Fixie Agent Python API Reference

@pydantic_dataclasses.dataclass
class AgentError:
    """An error that occurred in the process of generating a response."""

    # A code representing the error.
    code: str

    # A message describing the error that will be displayed to the user.
    message: str

    # Additional debugging context from the error.
    details: Optional[Dict]

@pydantic_dataclasses.dataclass
class AgentQuery:
    """A standalone query sent to a Fixie agent."""

    # The contents of the query.
    message: Message

    # This is an access token associated with the user for whom this query was
    # created. Agents wishing to make queries to other agents, or to other
    # Fixie services, should carry this token in the query so that it
    # can be tied back to the original user.
    access_token: Optional[str] = None

@pydantic_dataclasses.dataclass
class AgentResponse:
    """A response message from an Agent."""

    # The text of the response message.
    message: Message

    # Error details, if an error occurred generating the response. Note that `message` should still be included,
    # but may indicate that an error occurred that prevented the agent from fully handling the request.
    error: Optional[AgentError] = None

@pydantic_dataclasses.dataclass
class Embed:
    """An Embed represents a binary object attached to a Message."""

    # The MIME content type of the object, e.g., "image/png" or "application/json".
    content_type: str

    # A public URL where the object can be downloaded. This can be a data URI.
    uri: str

    @property
    def content(self) -> bytes:
        """Retrieves the content for this Embed object."""
        if self.uri.startswith("data:"):
            return base64.b64decode(self.uri.split(",")[1])
        return requests.get(self.uri).content

    @content.setter
    def content(self, content: bytes):
        """Sets the content of the Embed object as a data URI."""
        self.uri = f"data:base64,{base64.b64encode(content).decode('utf-8')}"

    @property
    def text(self) -> str:
        """Retrieves the content of the Embed object as a string."""
        return self.content.decode("utf-8")

    @text.setter
    def text(self, text: str):
        """Sets the content of the Embed object as a string."""
        self.content = text.encode("utf-8")

@pydantic_dataclasses.dataclass
class Message:
    """A Message represents a single message sent to a Fixie agent."""

    # The text of the message.
    text: str

    # A mapping of embed keys to Embed objects.
    embeds: Dict[str, Embed] = dataclasses.field(default_factory=dict)

class CodeShotAgent(agent_base.AgentBase):
    """A CodeShot agent.

    To make a CodeShot agent, simply pass a BASE_PROMPT and FEW_SHOTS:

        BASE_PROMPT = "A summary of what this agent does; how it does it; and its
        personality"

        FEW_SHOTS = '''
        Q: <Sample query that this agent supports>
        A: <Desired response for this query>

        Q: <Another sample query>
        A: <Desired response for this query>
        '''

        agent = CodeShotAgent(BASE_PROMPT, FEW_SHOTS)

    You can have FEW_SHOTS as a single string of all your few-shots separated by 2 new
    lines, or as an explicit list of one few-shot per index.

    Your few-shots may reach out to other Agents in the fixie ecosystem by
    "Ask Agent[agent_id]: <query to pass>", or reach out to some python functions
    by "Ask Func[func_name]: <query to pass>".

    There are a series of default runtime `Func`s provided by the platform available for
    your agents to consume. For a full list of default runtime `Func`s, refer to:
        http://docs.fixie.ai/XXX

    You may also need to write your own python functions here to be consumed by your
    agent. To make a function accessible by an agent, you'd need to register it by
    `@agent.register_func`. Example:

        @agent.register_func
        def func_name(query: fixieai.Message) -> ReturnType:
            ...

        , where ReturnType is one of `str`, `fixieai.Message`, or `fixie.AgentResponse`.

    Note that in the above, we are using the decorator `@agent.register_func` to
    register this function with the agent instance we just created.

    To check out the default `Func`s that are provided in Fixie, see:
        http://docs.fixie.ai/XXX

    """

    def __init__(
        self,
        base_prompt: str,
        few_shots: Union[str, List[str]],
        corpora: Optional[List[corpora.UrlDocumentCorpus]] = None,
        conversational: bool = False,
        oauth_params: Optional[oauth.OAuthParams] = None,
        llm_settings: Optional[llm_settings.LlmSettings] = None,
    ):
        super().__init__(oauth_params)

        if isinstance(few_shots, str):
            few_shots = _split_few_shots(few_shots)

        self.base_prompt = base_prompt
        self.few_shots = few_shots
        self.url_corpora = corpora
        self.conversational = conversational
        self.llm_settings = llm_settings

        utils.strip_prompt_lines(self)

    @property
    def corpora(self):
        url_corpora = self.url_corpora or []
        custom_corpora = [
            corpora.CustomCorpus(func_name=f) for f in self._corpus_funcs.keys()
        ]
        return url_corpora + custom_corpora if url_corpora or custom_corpora else None

    def metadata(self) -> metadata.Metadata:
        return metadata.CodeShotAgentMetadata(
            self.base_prompt,
            self.few_shots,
            self.corpora,
            self.conversational,
            self.llm_settings,
        )

    def validate(self):
        utils.validate_code_shot_agent(self)

@pydantic_dataclasses.dataclass
class CorpusDocument:
    """Some meaningful item of data from a corpus. This could be an HTML page,
    a PDF, or a raw string of text (among others). Fixie will handle parsing
    and chunking this document so that appropriately sized chunks can be
    included in LLM requests.

    Note: If custom parsing is desired, agents are free to implement their own
    parsing to return documents with text/plain mime_types instead of whatever
    they fetch natively. Fixie will not alter text/plain documents prior to
    chunking."""

    source_name: str
    content: bytes = dataclasses.field(
        metadata=dataclasses_json.config(
            encoder=lambda c: base64.b64encode(c).decode(),
            decoder=lambda c: base64.b64decode(c.encode()),
        )
    )
    encoding: str = "UTF-8"
    mime_type: str = "text/plain"

    @property
    def text(self) -> str:
        return self.content.decode(self.encoding)

@pydantic_dataclasses.dataclass
class CorpusPage:
    """A page of CorpusDocuments. In addition to the documents themselves, a
    page may include a continuation token for fetching the next page (in the
    same partition). Omitting a token implies that this is the last page."""

    documents: List[CorpusDocument]
    next_page_token: Optional[str] = None

@pydantic_dataclasses.dataclass
class CorpusPartition:
    """An identifier for a subset of a corpus, along with an optional
    token to use when loading its first page. Each partition will only be
    loaded once during a single crawl. If multiple responses include the same
    partition, the token of the first received response will be used."""

    partition: str
    first_page_token: Optional[str] = None

@pydantic_dataclasses.dataclass
class CorpusRequest(dataclasses_json.DataClassJsonMixin):
    """A request for some piece of the agent's corpus.

    In addition to returning documents, each response may expand the corpus
    space in one or both of two dimensions: new partitions and next pages.

    Partitions are non-overlapping subsets of a corpus which may be loaded in
    parallel by Fixie. A response's new partitions will be ignored if
    previously included in another response.

    When a response includes a page of documents, that page may indicate that
    another page is available in the same partition. Pages are always loaded
    serially in order. The partition is completed when a response has a page
    with no next_page_token.

    Agents will always receive a first request with the default (unnamed)
    partition and no page_token. Subsequent requests depend on prior responses
    and will always include at least one of those fields.

    Examples:
        Simple handful of documents:

            When receiving the initial request, the agent responds with a page
            of documents. This could include a next_page_token for more
            documents in the single default partition if needed.

        Web crawl:

            Each URL corresponds to a partition and the agent never returns
            tokens. The initial response only includes partitions, one for each
            root URL to crawl. Each subsequent request includes the partition
            (the URL) and the corresponding response contains a page with a
            single document - the resource at that URL. If the document links
            to other resources that should be included in the corpus, the
            response also contains those URLs as new partitions. The process
            repeats for all partitions until there are no known incomplete
            partitions or until crawl limits are reached.

        Database:

            Consider a database with a parent table keyed by parent_id and an
            interleaved child table keyed by (parent_id, child_id) whose rows
            correspond to corpus documents. This agent will use tokens that
            encode a read timestamp (for consistency) and an offset to be used
            in combination with a static page size.

            Upon receiving the initial CorpusRequest, the agent chooses a
            commit timestamp to use for all reads and returns a partition for
            each parent_id along with a first_page_token indicating the chosen
            read timestamp and an offset of 0.

            For each partition, the agent then receives requests with the
            partition (a parent_id) and a page token (the read timestamp and
            offest). It responds with documents corresponding to the next page
            size child rows within the given parent. If more children exist,
            the response includes a next_page_token with the same read
            timestamp and an incremented offset. This repeats until there are
            no more children, at which point the response has no
            next_page_token and the partition is complete.

            Note: Including multiple parent_ids in each partition would also
                work and would be an effective way to limit parallelism if
                desired.

    Attributes:
        partition: The partition of the corpus that should be read. This will
            be empty for the initial request, indicating the default partition.
            For subsequent requests, it will either be the name of a partition
            returned by a previous request or empty if the default partition
            contains multiple pages for this agent.
        page_token: A token for paginating results within a corpus partition.
            If present, this will be echoed from a previous response.
    """

    partition: Optional[str] = None
    page_token: Optional[str] = None

@pydantic_dataclasses.dataclass
class CorpusResponse(dataclasses_json.DataClassJsonMixin):
    """A response to a CorpusRequest. See CorpusRequest for details."""

    partitions: Optional[List[CorpusPartition]] = None
    page: Optional[CorpusPage] = None

@pydantic_dataclasses.dataclass
class CustomCorpus:
    """A custom corpus for a Fixie CodeShot Agent. This uses a registered
    corpus func to load documents from an arbitrary source."""

    func_name: str

@deprecated(reason="Use register_corpus_func for custom document loading.")
@pydantic_dataclasses.dataclass
class DocumentLoader:
    """Deprecated. This doesn't do anything."""

    name: str

@pydantic_dataclasses.dataclass
class UrlDocumentCorpus:
    """URL Document corpus for a Fixie CodeShot Agent."""

    urls: List[str]
    """URLs to load documents from. A trailing wildcard (e.g., https://example.com/*),
    can be used to load all documents from a site."""

    exclude_patterns: Optional[List[str]] = None
    """A list of wildcard patterns to exclude from crawled URLs (e.g., */no_crawl/*)."""

    auth_token_func: Optional[str] = None

    loader: Optional[DocumentLoader] = None  # Deprecated.

class OAuthHandler:
    """
    OAuthHandler that wraps around a (OAuthParams, query) to authenticate.

    This client object provides 3 main method:
        * credentials: Returns current user's OAuth access token, or None if they are
            not authenticated.
        * get_authorization_url: Returns a url that the users can click to
            authenticate themselves.
        * authorize: Exchanges a received access code (from auth redirect callback) for
            an access token. If successful, user's storage is updated.
    """

    # OAuth keys reserved in UserStorage
    OAUTH_STATE_KEY = "_oauth_state"
    OAUTH_TOKEN_KEY = "_oauth_token"

    def __init__(
        self,
        oauth_params: OAuthParams,
        token_claims: token.VerifiedTokenClaims,
    ):
        self._storage = user_storage.UserStorage(token_claims)
        self._oauth_params = oauth_params
        self._agent_id = token_claims.agent_id

    def get_authorization_url(self) -> str:
        """Returns a URL to launch the authorization flow."""
        auth_state = f"{self._agent_id}:{secrets.token_urlsafe()}"
        data = {
            "response_type": "code",
            "access_type": "offline",
            "client_id": self._oauth_params.client_id,
            "scope": " ".join(self._oauth_params.scopes),
            "state": auth_state,
            "redirect_uri": constants.FIXIE_OAUTH_REDIRECT_URL,
        }
        url = self._oauth_params.auth_uri + "?" + parse.urlencode(data)
        # Store auth_state in UserStorage for validation later.
        self._storage[self.OAUTH_STATE_KEY] = auth_state
        return url

    def user_token(self) -> Optional[str]:
        """Returns current user's OAuth credentials, or None if not authorized."""
        try:
            creds_json = self._storage[self.OAUTH_TOKEN_KEY]
        except KeyError:
            return None

        if not isinstance(creds_json, str):
            logging.warning(
                f"Value at user_storage[{self.OAUTH_TOKEN_KEY!r}] is "
                f"not an OAuthCredentials json: {creds_json!r}"
            )
            del self._storage[self.OAUTH_TOKEN_KEY]
            return None

        try:
            creds = _OAuthCredentials.from_json(creds_json)
        except (TypeError, LookupError, ValueError):
            logging.warning(
                f"Value at user_storage[{self.OAUTH_TOKEN_KEY!r}] is "
                f"not a valid _OAuthCredentials json: {creds_json!r}"
            )
            del self._storage[self.OAUTH_TOKEN_KEY]
            return None

        if creds.expired:
            logging.debug(f"Credentials expired at {creds.expiry}")
            if not creds.refresh_token:
                logging.warning("No refresh token available")
                return None
            logging.debug("Refreshing credentials...")
            creds.refresh(self._oauth_params)
            self._save_credentials(creds)  # Save refreshed token to UserStorage
        return creds.access_token

    def authorize(self, state: str, code: str):
        """Authorize the received access `code` against the client secret.

        If successful, the credentials will be saved in user storage.
        """
        expected_state = self._storage[self.OAUTH_STATE_KEY]
        if state != expected_state:
            logging.warning(
                f"Unknown state token, expected: {expected_state!r} actual: {state!r}"
            )
            raise ValueError(f"Unknown state token")

        data = {
            "grant_type": "authorization_code",
            "client_id": self._oauth_params.client_id,
            "client_secret": self._oauth_params.client_secret,
            "code": code,
            "redirect_uri": constants.FIXIE_OAUTH_REDIRECT_URL,
        }
        response = _send_authorize_request(self._oauth_params.token_uri, data)
        logging.debug(
            f"OAuth auth request succeeded, lifetime={response.expires_in} "
            f"refreshable={response.refresh_token is not None}"
        )
        credentials = _OAuthCredentials(
            response.access_token,
            _get_expiry(response.expires_in),
            response.refresh_token,
        )
        self._save_credentials(credentials)

    def _save_credentials(self, credentials: "_OAuthCredentials"):
        self._storage[self.OAUTH_TOKEN_KEY] = credentials.to_json()

@dataclasses.dataclass
class OAuthParams:
    """Encapsulates OAuth parameters, including secret, auth uri, and the scope.

    Agents who want to use OAuth flow, should declare their secrets via this object.
    """

    client_id: str
    client_secret: str
    auth_uri: str
    token_uri: str
    scopes: List[str]

    @classmethod
    def from_client_secrets_file(
        cls, secrets_path: str, scopes: List[str]
    ) -> "OAuthParams":
        """Initializes OAuth from a secrets file, e.g., as obtained from the Google
        Cloud Console.

        Args:
            secrets_path: Path to a json file holding secret values.
            scopes: A list of scopes that access needs to be requested for.
        """
        with open(secrets_path, "r") as file:
            data = json.load(file)
            secrets = data["web"] or data["installed"]
            return cls(
                secrets["client_id"],
                secrets["client_secret"],
                secrets["auth_uri"],
                secrets["token_uri"],
                scopes,
            )

class StandaloneAgent(agent_base.AgentBase):
    """An agent that handles queries directly.

    To make a StandaloneAgent, pass a function with the following signature

    def handle(query: fixieai.Message) -> ReturnType:
            ...

    where ReturnType is one of `str`, `fixieai.Message`, or `fixie.AgentResponse`.
    """

    def __init__(
        self,
        handle_message: Callable,
        sample_queries: Optional[List[str]] = None,
        oauth_params: Optional[oauth.OAuthParams] = None,
    ):
        super().__init__(oauth_params)

        if isinstance(handle_message, agent_func.AgentFunc):
            self._handle_message: agent_func.AgentFunc = handle_message
        else:
            self._handle_message = agent_func.AgentQueryFunc.create(
                handle_message,
                oauth_params,
                default_message_type=api.Message,
                allow_generator=True,
            )
        self._sample_queries = sample_queries

    def metadata(self) -> metadata.Metadata:
        return metadata.StandaloneAgentMetadata(sample_queries=self._sample_queries)

    def validate(self):
        pass

    def api_router(self) -> fastapi.APIRouter:
        router = super().api_router()
        router.add_api_route("/", self._serve_query, methods=["POST"])
        return router

    def _serve_query(
        self,
        query: api.AgentQuery,
        credentials: fastapi.security.HTTPAuthorizationCredentials = fastapi.Depends(
            fastapi.security.HTTPBearer()
        ),
    ) -> fastapi.responses.Response:
        """Verifies the request is a valid request from Fixie, and dispatches it to
        the previously specified `handle_message` function. Depending on the return
        value of that function, either a single or a streaming response is returned.
        """
        token_claims = super()._check_credentials(credentials)

        output = self._handle_message(query, token_claims)
        return fastapi.responses.StreamingResponse(
            (json.dumps(dataclasses.asdict(resp)) + "\n" for resp in output),
            media_type="application/json",
        )

class UserStorage(MutableMapping[str, UserStorageType]):
    """UserStorage provides a dict-like interface to a user-specific storage.

    Usage:
    >>> from fixieai.agents import token
    >>> storage = UserStorage(token.VerifiedTokenClaims("fake-agent", False, "fake-access-token"))
    >>> storage["key"] = "value"
    >>> storage["complex-key"] = {"key1": {"key2": [12, False, None, b"binary"]}}
    >>> assert len(storage) == 2
    >>> assert storage["complex-key"]["key1"]["key2"][-1] == b"binary"
    """

    def __init__(
        self,
        token_claims: token.VerifiedTokenClaims,
        userstorage_url: str = constants.FIXIE_USER_STORAGE_URL,
    ):
        if token_claims.is_anonymous:
            raise exceptions.AgentException(
                response_message="I'm sorry, you must login to use this agent.",
                error_code="ERR_USERSTORAGE_REQUIRES_USER",
                error_message="This agent requires user storage, which is not available to anonymous users. Please login or create an account.",
                http_status_code=400,
            )

        self._agent_id = token_claims.agent_id
        self._userstorage_url = userstorage_url
        self._session = requests.Session()
        self._session.headers.update({"Authorization": f"Bearer {token_claims.token}"})

    def __setitem__(self, key: str, value: UserStorageType):
        url = f"{self._userstorage_url}/{self._agent_id}/{key}"
        response = self._session.post(url, json={"data": to_json(value)})
        response.raise_for_status()

    def __getitem__(self, key: str) -> UserStorageType:
        url = f"{self._userstorage_url}/{self._agent_id}/{key}"
        try:
            response = self._session.get(url)
            response.raise_for_status()
            return from_json(response.json()["data"])
        except requests.exceptions.HTTPError as e:
            raise KeyError(f"Key {key} not found") from e

    def __contains__(self, key: object) -> bool:
        url = f"{self._userstorage_url}/{self._agent_id}/{key}"
        try:
            response = self._session.head(url)
            response.raise_for_status()
            return True
        except requests.exceptions.HTTPError as e:
            return False

    def __delitem__(self, key: str):
        url = f"{self._userstorage_url}/{self._agent_id}/{key}"
        try:
            response = self._session.delete(url)
            response.raise_for_status()
        except requests.exceptions.HTTPError as e:
            raise KeyError(f"Key {key} not found") from e

    def _get_all_keys(self):
        url = f"{self._userstorage_url}/{self._agent_id}"
        response = self._session.get(url)
        response.raise_for_status()
        return [value["key"] for value in response.json()]

    def __iter__(self):
        return iter(self._get_all_keys())

    def __len__(self):
        return len(self._get_all_keys())

class FewshotLinePattern(enum.Enum):
    QUERY = re.compile(r"^Q:")
    AGENT_SAYS = re.compile(r"^Agent\[(?P<agent_id>\w+)] says:")
    FUNC_SAYS = re.compile(r"^Func\[(?P<func_name>\w+)] says:")
    ASK_AGENT = re.compile(r"^Ask Agent\[(?P<agent_id>\w+)]:")
    ASK_FUNC = re.compile(r"^Ask Func\[(?P<func_name>\w+)]:")
    RESPONSE = re.compile(r"^A:")
    NO_PATTERN: None = None

    @classmethod
    def match(cls, line: str) -> Optional[re.Match[str]]:
        """Returns a match from a FewshotLinePattern for a given line, or None if nothing matched."""
        if "\n" in line:
            raise ValueError(
                "Cannot get the pattern for a multi-line text. Patterns must be "
                "extracted one line at a time."
            )
        matches = [
            match
            for prompt_pattern in cls
            if prompt_pattern is not cls.NO_PATTERN
            and (match := prompt_pattern.value.match(line))
        ]
        if len(matches) > 1:
            raise RuntimeError(
                f"More than one pattern ({list(FewshotLinePattern(match.re) for match in matches)}) matched the line {line!r}."
            )
        elif len(matches) == 0:
            return None

        match = matches[0]
        if match.re is cls.QUERY.value:
            if match.end() == len(line):
                raise ValueError("A 'Q:' line cannot end without a query.")

        assert isinstance(match, re.Match)
        return match