Fixie Client Python API Reference

class FixieClient:
    """FixieClient is a client to the Fixie system.

    Args:
        api_key: The API key for the Fixie API server. If not provided, the
            FIXIE_API_KEY environment variable will be used. If that is not
            set, the authenticated user API key will be used, or a ValueError
            will be raised if the user is not authenticated.
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
    ):
        self._api_key = api_key or constants.fixie_api_key()
        logging.info(f"Using Fixie API URL: {constants.FIXIE_API_URL}")
        self._request_headers = {"Authorization": f"Bearer {self._api_key}"}
        transport = RequestsHTTPTransport(
            url=constants.FIXIE_GRAPHQL_URL,
            headers=self._request_headers,
        )
        self._gqlclient = Client(transport=transport, fetch_schema_from_transport=False)

    @property
    def gqlclient(self) -> Client:
        """Return the underlying GraphQL client used by this FixieClient."""
        return self._gqlclient

    @property
    def url(self) -> str:
        """Return the URL of the Fixie API server."""
        return constants.FIXIE_API_URL

    def clone(self) -> "FixieClient":
        """Return a new FixieClient instance with the same configuration."""
        return FixieClient(api_key=self._api_key)

    def get_agents(self) -> Dict[str, Dict[str, str]]:
        """Return metadata about all running Fixie Agents. The keys of the returned
        dictionary are the Agent handles, and the values are dictionaries containing
        metadata about each Agent."""

        query = gql(
            """
            query getAgents {
                allAgents {
                    agentId
                    name
                    description
                    moreInfoUrl
                }
            }
        """
        )
        result = self._gqlclient.execute(query)
        assert "allAgents" in result and isinstance(result["allAgents"], list)
        agents = result["allAgents"]
        return {agent["agentId"]: agent for agent in agents}

    def get_agent(self, agent_id: str) -> Agent:
        """Return an existing Agent object."""
        return Agent(self, agent_id)

    def get_agent_page_url(self, agent_id: str) -> str:
        """Return the URL of the Agent page on the Fixie Platform."""
        return f"{constants.FIXIE_AGENT_URL}/{agent_id}"

    def create_agent(
        self,
        handle: str,
        name: str,
        description: str,
        query_url: Optional[str] = None,
        func_url: Optional[str] = None,
        more_info_url: Optional[str] = None,
        published: Optional[bool] = None,
    ) -> Agent:
        """Create a new Agent.

        Args:
            handle: The handle for the new Agent. This must be unique across all
                Agents owned by this user.
            name: The name of the new Agent.
            description: A description of the new Agent.
            query_url: (deprecated) Unused.
            func_url: (deprecated) The URL of the new Agent's func endpoint.
            more_info_url: A URL with more information about the new Agent.
            published: Whether the new Agent should be published.
        """
        agent = Agent(self, f"{self.get_current_username()}/{handle}")
        agent.create_agent(
            name, description, query_url, func_url, more_info_url, published
        )
        return agent

    def get_sessions(self) -> List[str]:
        """Return a list of all session IDs."""

        query = gql(
            """
            query getSessions {
                allSessions {
                    handle
                }
            }
        """
        )
        result = self._gqlclient.execute(query)
        assert "allSessions" in result and isinstance(result["allSessions"], list)
        sessions = result["allSessions"]
        return [session["handle"] for session in sessions]

    def create_session(self, frontend_agent_id: Optional[str] = None) -> Session:
        """Create a new Session."""
        return Session(self, frontend_agent_id=frontend_agent_id)

    def get_session(self, session_id: str) -> Session:
        """Return an existing Session object."""
        return Session(self, session_id)

    def get_current_user(self) -> Dict[str, Any]:
        """Returns the username of the current user."""
        query = gql(
            """
            query getUsername {
                user {
                    username
                    firstName
                    lastName
                    email
                    organization {
                        name
                    }
                    dailyQueryLimit
                    dailyUsedQueries
                    avatar
                }
            }
        """
        )
        result = self._gqlclient.execute(query)
        assert "user" in result and isinstance(result["user"], dict)
        return result["user"]

    def get_current_username(self) -> str:
        """Returns the username of the current user."""
        user = self.get_current_user()
        assert "username" in user and isinstance(user["username"], str)
        return user["username"]

    def refresh_agent(self, agent_handle: str):
        """Indicates that an agent's prompts should be refreshed."""
        username = self.get_current_username()
        requests.post(
            f"{constants.FIXIE_REFRESH_URL}/{username}/{agent_handle}",
            headers=self._request_headers,
        ).raise_for_status()

    def create_agent_revision(
        self,
        handle: str,
        make_current: bool,
        *,
        reindex_corpora: bool = False,
        metadata: Optional[Dict[str, str]] = None,
        external_url: Optional[str] = None,
        python_gzip_tarfile: Optional[BinaryIO] = None,
    ) -> str:
        """Creates a new Agent revision.

        Args:
            handle: The handle of the Agent. Must be owned by the current user.
            make_current: Whether the new revision should be made the current (active) revision.
            reindex_corpora: Whether to reindex all corpora for the new revision.
            metadata: Optional client-provided metadata to associate with the revision.
            external_url: The URL at which the revision is hosted, if hosted externally.
            python_gzip_tarfile: A file-like of a gzip-compressed tarfile containing the files to deploy.

        Exactly one of `external_url` and `python_gzip_tarfile` must be provided.
        """

        mutation = gql(
            """
            mutation CreateAgentRevision(
                $handle: String!,
                $metadata: [RevisionMetadataKeyValuePairInput!]!,
                $makeCurrent: Boolean!,
                $externalDeployment: ExternalDeploymentInput,
                $managedDeployment: ManagedDeploymentInput,
                $reindexCorpora: Boolean!,
            ) {
                createAgentRevision(agentHandle: $handle, makeCurrent: $makeCurrent, revision: { 
                    metadata: $metadata
                    externalDeployment: $externalDeployment
                    managedDeployment: $managedDeployment
                }, reindexCorpora: $reindexCorpora) {
                    revision {
                        id
                    }
                }
            }
            """
        )

        with utils.patched_gql_file_uploader(
            python_gzip_tarfile, "upload.tar.gz", "application/gzip"
        ):
            result = self._gqlclient.execute(
                mutation,
                variable_values={
                    "handle": handle,
                    "metadata": [
                        {"key": key, "value": value} for key, value in metadata.items()
                    ]
                    if metadata
                    else [],
                    "makeCurrent": make_current,
                    "reindexCorpora": reindex_corpora,
                    "externalDeployment": {"url": external_url}
                    if external_url
                    else None,
                    "managedDeployment": {
                        "environment": "PYTHON",
                        "codePackage": python_gzip_tarfile,
                    }
                    if python_gzip_tarfile
                    else None,
                },
                upload_files=True,
            )

        revision_id = result["createAgentRevision"]["revision"]["id"]
        assert isinstance(revision_id, str)
        return revision_id

    def get_current_agent_revision(self, handle: str) -> Optional[str]:
        """Gets the current (active) revision of an agent."""

        query = gql(
            """
            query GetRevisionId($handle: String!) {
                agentByHandle(handle: $handle) {
                    currentRevision {
                        id
                    }
                }
            }
            """
        )

        response = self._gqlclient.execute(
            query,
            variable_values={"handle": handle},
        )

        revision = response["agentByHandle"]["currentRevision"]
        return revision["id"] if revision is not None else None

    def set_current_agent_revision(self, handle: str, revision_id: str):
        """Sets the current (active) revision of an agent."""

        mutation = gql(
            """
            mutation SetCurrentAgentRevision(
                $handle: String!,
                $currentRevisionId: ID!) {
                updateAgent(
                    agentData: {
                        handle: $handle,
                        currentRevisionId: $currentRevisionId
                    }
                ) {
                    agent {
                        agentId
                    }
                }
            }
            """
        )

        _ = self._gqlclient.execute(
            mutation,
            variable_values={"handle": handle, "currentRevisionId": revision_id},
        )

    def delete_agent_revision(self, handle: str, revision_id: str):
        """Deletes an Agent revision."""

        mutation = gql(
            """
            mutation DeleteRevision($handle: String!, $revisionId: ID!) {
                deleteAgentRevision(agentHandle: $handle, revisionId: $revisionId) {
                    agent {
                        agentId
                    }
                }
            }
            """
        )
        _ = self._gqlclient.execute(
            mutation, variable_values={"handle": handle, "revisionId": revision_id}
        )

    def deploy_agent(
        self,
        handle: str,
        gzip_tarfile: BinaryIO,
    ):
        """Deploys an agent implementation.

        Args:
            handle: The handle of the Agent to deploy. Must be owned by the current user.
            gzip_tarfile: A file-like of a gzip-compressed tarfile containing the files to deploy.
        """
        return self.create_agent_revision(
            handle,
            make_current=True,
            python_gzip_tarfile=gzip_tarfile,
        )

class Session:
    """Represents a single session with the Fixie system.

    Args:
        client: The FixieClient instance to use.
        session_id: The ID of the session to use. If not provided, a new
            session will be created.
    """

    def __init__(
        self,
        client: FixieClient,
        session_id: Optional[str] = None,
        frontend_agent_id: Optional[str] = None,
    ):
        self._client = client
        self._gqlclient = self._client.gqlclient
        self._session_id = session_id
        if session_id:
            # Test that the session exists.
            _ = self.get_metadata()
            assert (
                frontend_agent_id is None
            ), "Cannot specify frontend_agent_id when using an existing session"
        else:
            self._session_id = self._create_session(frontend_agent_id)
        self._frontend_agent_id: Optional[str] = None
        self._last_message_timestamp: Optional[datetime.datetime] = None

    @property
    def session_id(self) -> Optional[str]:
        """Return the session ID used by this Fixie client."""
        return self._session_id

    @property
    def session_url(self) -> str:
        """Return the URL of the Fixie session."""
        return f"{self._client.url}/sessions/{self.session_id}"

    @property
    def frontend_agent_id(self) -> Optional[str]:
        """Return the frontend agent ID used by this Fixie client."""
        return self._frontend_agent_id

    def clone(self) -> "Session":
        """Return a new Session instance with the same configuration."""
        return Session(self._client.clone(), session_id=self._session_id)

    def _create_session(self, frontend_agent_id: Optional[str] = None) -> str:
        """Create a new session."""
        assert self._session_id is None

        query = gql(
            """
            mutation CreateSession($frontendAgentId: String) {
                createSession(sessionData: {frontendAgentId: $frontendAgentId}) {
                    session {
                        handle
                        frontendAgentId
                    }
                }
            }
            """
        )

        result = self._gqlclient.execute(
            query, variable_values={"frontendAgentId": frontend_agent_id}
        )
        if "createSession" not in result or result["createSession"] is None:
            raise ValueError(f"Failed to create Session")
        assert isinstance(result["createSession"], dict)
        assert isinstance(result["createSession"]["session"], dict)
        self._frontend_agent_id = result["createSession"]["session"]["frontendAgentId"]
        assert isinstance(result["createSession"]["session"]["handle"], str)
        return result["createSession"]["session"]["handle"]

    def get_metadata(self) -> Dict[str, Any]:
        """Return metadata about this session."""

        query = gql(
            """
            query getSession($session_id: String!) {
                sessionByHandle(handle: $session_id) {
                    handle
                    name
                    description
                    frontendAgentId
                }
            }
        """
        )
        result = self._gqlclient.execute(
            query, variable_values={"session_id": self._session_id}
        )
        assert "sessionByHandle" in result and isinstance(
            result["sessionByHandle"], dict
        )
        self._frontend_agent_id = result["sessionByHandle"]["frontendAgentId"]
        return result["sessionByHandle"]

    def delete_session(self) -> None:
        """Delete the current session."""
        query = gql(
            """
            mutation DeleteSession($handle: String!) {
                deleteSession(handle: $handle) {
                    session {
                        handle
                    }
                }
            }
        """
        )
        _ = self._gqlclient.execute(query, variable_values={"handle": self._session_id})

    def get_embeds(self) -> List[Dict[str, Any]]:
        """Return the Embeds attached to this Session."""
        query = gql(
            """
            query getEmbeds($handle: String!) {
                sessionByHandle(handle: $handle) {
                    embeds {
                        key
                        embed {
                            id
                            contentType
                            created
                            contentHash
                            owner {
                                username
                            }
                            url
                        }
                    }
                }
            }
        """
        )
        result = self._gqlclient.execute(
            query, variable_values={"handle": self._session_id}
        )
        assert (
            "sessionByHandle" in result
            and isinstance(result["sessionByHandle"], dict)
            and isinstance(result["sessionByHandle"]["embeds"], list)
        )
        return result["sessionByHandle"]["embeds"]

    def get_messages(self) -> List[Dict[str, Any]]:
        """Return the messages that make up this session."""
        query = gql(
            """
            query getMessages($handle: String!) {
                sessionByHandle(handle: $handle) {
                    messages {
                        id
                        text
                        sentBy {
                            handle
                        }
                        type
                        inReplyTo { id }
                        timestamp
                    }
                }
            }
        """
        )
        result = self._gqlclient.execute(
            query, variable_values={"handle": self._session_id}
        )
        assert (
            "sessionByHandle" in result
            and isinstance(result["sessionByHandle"], dict)
            and isinstance(result["sessionByHandle"]["messages"], list)
        )
        return result["sessionByHandle"]["messages"]

    def add_message(self, text: str) -> str:
        """Add a message to this Session. Returns the added message text."""
        query = gql(
            """
            mutation Post($handle: String!, $text: String!) {
                sendSessionMessage(messageData: {session: $handle, text: $text}) {
                    message {
                        text
                    }
                }
            }
            """
        )
        result = self._gqlclient.execute(
            query, variable_values={"handle": self._session_id, "text": text}
        )
        assert isinstance(result["sendSessionMessage"]["message"]["text"], str)
        return result["sendSessionMessage"]["message"]["text"]

    def query(self, text: str) -> str:
        """Run a single query against the Fixie API and return the response."""
        self.add_message(text)
        # The reply to the query comes in as the most recent 'response' message in the
        # session.
        response = self.get_messages()[-1]
        assert isinstance(response["text"], str)
        return response["text"]

    def run(self, text: str) -> Generator[Dict[str, Any], None, None]:
        """Run a query against the Fixie API, returning a generator that yields
        messages."""

        # Run the query in the background, and continue polling for replies.
        background_client = self.clone()
        threading.Thread(target=background_client.add_message, args=(text,)).start()

        response_received = False
        while not response_received:
            time.sleep(1)
            messages = self.get_messages_since_last_time()
            for message in messages:
                response_received = message["type"] == "response"
                yield message

    def get_messages_since_last_time(self) -> List[Dict[str, Any]]:
        """Return all messages since the given timestamp."""
        timestamp = self._last_message_timestamp
        messages_since_last_time = []
        for message in self.get_messages():
            message_timestamp = datetime.datetime.fromisoformat(message["timestamp"])
            if timestamp is None or message_timestamp > timestamp:
                messages_since_last_time.append(message)
                self._last_message_timestamp = message_timestamp
        return messages_since_last_time

class Agent:
    """Provides an interface to the Fixie GraphQL Agent API.

    Args:
        client: The FixieClient instance to use.
        agent_id: The Agent ID, e.g., "fixie/calc", or handle, e.g., "dice".
    """

    def __init__(
        self,
        client: FixieClient,
        agent_id: str,
    ):
        self._client = client
        self._gqlclient = self._client.gqlclient

        self._agent_id = agent_id
        self._owner: Optional[str] = None
        if "/" in agent_id:
            self._owner, self._handle = agent_id.split("/")
        else:
            self._handle = agent_id

        self._metadata: Optional[Dict[str, Any]] = None
        try:
            self._metadata = self.get_metadata()
        except:
            self._metadata = None

    @property
    def agent_id(self) -> str:
        """Return the agentId for this Agent."""
        return self._agent_id

    @property
    def handle(self) -> str:
        """Return the handle for this Agent."""
        return self._handle

    @property
    def valid(self) -> bool:
        """Return whether this Agent is valid."""
        return self._metadata is not None

    @property
    def name(self) -> Optional[str]:
        """Return the name for this Agent."""
        if self._metadata is None:
            return None
        name = self._metadata["name"]
        assert name is None or isinstance(name, str)
        return name

    @property
    def description(self) -> Optional[str]:
        """Return the description for this Agent."""
        if self._metadata is None:
            return None
        description = self._metadata["description"]
        assert description is None or isinstance(description, str)
        return description

    @property
    def queries(self) -> Optional[List[str]]:
        """Return the queries for this Agent."""
        if self._metadata is None:
            return None
        queries = self._metadata["queries"]
        assert queries is None or (
            isinstance(queries, list) and all(isinstance(q, str) for q in queries)
        )
        return queries

    @property
    def more_info_url(self) -> Optional[str]:
        """Return the more info URL for this Agent."""
        if self._metadata is None:
            return None
        more_info_url = self._metadata["moreInfoUrl"]
        assert more_info_url is None or isinstance(more_info_url, str)
        return more_info_url

    @property
    def published(self) -> Optional[bool]:
        """Return the published status for this Agent."""
        if self._metadata is None:
            return None
        published = self._metadata["published"]
        assert published is None or isinstance(published, bool)
        return published

    @property
    def owner(self) -> Optional[str]:
        """Return the owner of this Agent."""
        if self._metadata is None:
            return None
        owner_username = self._metadata["owner"]["username"]
        assert owner_username is None or isinstance(owner_username, str)
        return owner_username

    @property
    def query_url(self) -> Optional[str]:
        """Return the query URL for this Agent."""
        if self._metadata is None:
            return None
        url = self._metadata["queryUrl"]
        assert url is None or isinstance(url, str)
        return url

    @property
    def func_url(self) -> Optional[str]:
        """Return the func URL for this Agent."""
        if self._metadata is None:
            return None
        url = self._metadata["funcUrl"]
        assert url is None or isinstance(url, str)
        return url

    @property
    def created(self) -> Optional[datetime.datetime]:
        """Return the creation timestamp for this Agent."""
        if self._metadata is None:
            return None
        ts = self._metadata["created"]
        if ts is not None:
            return datetime.datetime.fromisoformat(ts)
        else:
            return None

    @property
    def modified(self) -> Optional[datetime.datetime]:
        """Return the modification timestamp for this Agent."""
        if self._metadata is None:
            return None
        ts = self._metadata["modified"]
        if ts is not None:
            return datetime.datetime.fromisoformat(ts)
        else:
            return None

    def get_metadata(self) -> Dict[str, Any]:
        """Return metadata about this Agent."""

        if self._owner is None:
            # Query by handle.
            query = gql(
                """
                query getAgentByHandle($handle: String!) {
                    agentByHandle(handle: $handle) {
                        agentId
                        handle
                        name
                        description
                        queries
                        moreInfoUrl
                        published
                        owner {
                            username
                        }
                        queryUrl
                        funcUrl
                        created
                        modified
                    }
                }
            """
            )
            result = self._gqlclient.execute(
                query, variable_values={"handle": self._handle}
            )
            if "agentByHandle" not in result or result["agentByHandle"] is None:
                raise ValueError(f"Cannot fetch agent metadata for {self._handle}")
            agent_dict = result["agentByHandle"]

        else:
            # Query by agent ID.
            query = gql(
                """
                query getAgentById($agentId: String!) {
                    agentById(agentId: $agentId) {
                        agentId
                        handle
                        name
                        description
                        queries
                        moreInfoUrl
                        published
                        owner {
                            username
                        }
                        queryUrl
                        funcUrl
                        created
                        modified
                    }
                }
            """
            )
            result = self._gqlclient.execute(
                query, variable_values={"agentId": f"{self._owner}/{self._handle}"}
            )
            if "agentById" not in result or result["agentById"] is None:
                raise ValueError(
                    f"Cannot fetch agent metadata for {self._owner}/{self._handle}"
                )
            agent_dict = result["agentById"]

        assert isinstance(agent_dict, dict) and all(
            isinstance(k, str) for k in agent_dict.keys()
        )
        return agent_dict

    def create_agent(
        self,
        name: Optional[str],
        description: str,
        query_url: Optional[str] = None,
        func_url: Optional[str] = None,
        more_info_url: Optional[str] = None,
        published: Optional[bool] = None,
    ) -> str:
        """Create a new Agent with the given parameters."""
        query = gql(
            """
            mutation CreateAgent(
                $handle: String!,
                $name: String,
                $description: String!,
                $queryUrl: String,
                $funcUrl: String,
                $moreInfoUrl: String,
                $published: Boolean) {
                createAgent(
                    agentData: {
                        handle: $handle,
                        name: $name,
                        description: $description,
                        queryUrl: $queryUrl,
                        funcUrl: $funcUrl,
                        moreInfoUrl: $moreInfoUrl,
                        published: $published
                    }
                ) {
                    agent {
                        agentId
                    }
                }
            }
            """
        )

        variable_values: Dict[str, Any] = {"handle": self._handle}
        if name is not None:
            variable_values["name"] = name
        variable_values["description"] = description
        if query_url is not None:
            variable_values["queryUrl"] = query_url
        if func_url is not None:
            warnings.warn(
                "Setting func_url via create_agent is deprecated, use FixieClient.create_agent_revision instead."
            )
            variable_values["funcUrl"] = func_url
        if more_info_url is not None:
            variable_values["moreInfoUrl"] = more_info_url
        if published is not None:
            variable_values["published"] = published

        result = self._gqlclient.execute(query, variable_values=variable_values)
        if "createAgent" not in result or result["createAgent"] is None:
            raise ValueError(f"Failed to create Agent")
        assert isinstance(result["createAgent"], dict)
        assert isinstance(result["createAgent"]["agent"], dict)
        assert isinstance(result["createAgent"]["agent"]["agentId"], str)
        self._metadata = self.get_metadata()
        return result["createAgent"]["agent"]["agentId"]

    def update_agent(
        self,
        new_handle: Optional[str] = None,
        name: Optional[str] = None,
        description: Optional[str] = None,
        query_url: Optional[str] = None,
        func_url: Optional[str] = None,
        more_info_url: Optional[str] = None,
        published: Optional[bool] = None,
    ) -> str:
        """Update the Agent with the given parameters."""
        query = gql(
            """
            mutation UpdateAgent(
                $handle: String!,
                $newHandle: String,
                $name: String,
                $description: String,
                $queryUrl: String,
                $funcUrl: String,
                $moreInfoUrl: String,
                $published: Boolean) {
                updateAgent(
                    agentData: {
                        handle: $handle,
                        newHandle: $newHandle,
                        name: $name,
                        description: $description,
                        queryUrl: $queryUrl,
                        funcUrl: $funcUrl,
                        moreInfoUrl: $moreInfoUrl,
                        published: $published
                    }
                ) {
                    agent {
                        agentId
                    }
                }
            }
            """
        )

        variable_values: Dict[str, Any] = {"handle": self._handle}
        if new_handle is not None:
            variable_values["newHandle"] = new_handle
        if name is not None:
            variable_values["name"] = name
        if description is not None:
            variable_values["description"] = description
        if query_url is not None:
            variable_values["queryUrl"] = query_url
        if func_url is not None:
            warnings.warn(
                "Setting func_url via update_agent is deprecated, use FixieClient.create_agent_revision instead."
            )
            variable_values["funcUrl"] = func_url
        if more_info_url is not None:
            variable_values["moreInfoUrl"] = more_info_url
        if published is not None:
            variable_values["published"] = published

        result = self._gqlclient.execute(query, variable_values=variable_values)
        if "updateAgent" not in result or result["updateAgent"] is None:
            raise ValueError(f"Failed to update Agent")
        assert isinstance(result["updateAgent"], dict)
        assert isinstance(result["updateAgent"]["agent"], dict)
        assert isinstance(result["updateAgent"]["agent"]["agentId"], str)
        self._metadata = self.get_metadata()
        if new_handle:
            self._handle = new_handle
        return result["updateAgent"]["agent"]["agentId"]

    def delete_agent(self) -> None:
        """Delete this Agent."""
        query = gql(
            """
            mutation DeleteAgent($handle: String!) {
                deleteAgent(handle: $handle) {
                    agent {
                        handle
                    }
                }
            }
        """
        )
        _ = self._gqlclient.execute(query, variable_values={"handle": self._handle})