# Qualcomm AI MCP Server and Query Tool

This guide shows how to create a small [Model Context Protocol (MCP)](https://modelcontextprotocol.io/) server with a query tool that calls the agent query endpoint through the Apigee gateway and a refresh tool that renews its access token through the portal.

The example uses Python and the MCP Python SDK. Access and refresh tokens are read at runtime, and the MCP server refreshes the access token through the token portal when needed. Tokens are not stored in source code or written to ordinary files.

## What the tool does

The `query_contextual_agent` tool accepts a user message plus optional conversation, retrieval, filtering, and generation settings. A basic call still needs only `message` and sends this request body to the gateway:

```http
POST https://dev.apigwx-op.qualcomm.com/ce/v1/contextual-ai-harness/agents/d6c12230-018a-4b99-af1b-e0bed13dd49a/query/acl
Authorization: Bearer <production-issued-token>
Contextual-Workspace: dragon
Content-Type: application/json
Accept: application/json

{"messages":[{"role":"user","content":"Hello"}],"stream":false}
```

The tool returns the parsed JSON response from the API, including the agent message and any available retrievals or attributions. The separate `refresh_contextual_token` tool renews the access token through the portal; the refresh token is never sent to the gateway.

The inputs below follow the local Qualcomm AI Query API reference available in the portal's API Reference modal. Its typed inputs let MCP hosts discover the available options without depending on an external documentation site.

The example forwards these options through the Qualcomm Apigee gateway. Availability depends on the deployed gateway and agent; the public reference does not verify that every option is enabled there. This tool returns one JSON result and sets `stream` to `false`; it does not implement the API's streaming response mode.

## Query options

All inputs except `message` are optional.

| Tool input | Purpose and API mapping |
| --- | --- |
| `message` | Latest user question. Appended to the API's `messages` array with `role: "user"`. |
| `agent_id` | Agent UUID in the request path. Defaults to `QUALCOMM_AI_AGENT_ID`, or the supplied agent when unset. |
| `conversation_id` | Continue a conversation using the UUID returned by a previous query to the same agent. |
| `history` | Earlier messages in chronological order, each with `role`, `content`, and optional `custom_tags`. Supported roles: `user`, `assistant`, `system`, `knowledge`. The latest `message` is appended automatically. |
| `custom_tags` | String tags attached to the latest user message. |
| `retrievals_only` | URL query parameter. Set to `true` to retrieve sources without generating an answer. Defaults to `false`. |
| `include_retrieval_content_text` | URL query parameter. Set to `true` to include retrieved text with the answer. Defaults to `false`; retrieval-only mode includes source text automatically. |
| `llm_model_id` | Specific fine-tuned or aligned LLM model ID, when available to the agent. |
| `structured_output` | Object containing `json_schema` and optional `type: "JSON"` to request a structured answer. |
| `documents_filters` | A metadata filter or nested logical group of filters. See examples below. |
| `override_configuration` | Per-query agent settings, using the documented fields below. Omitted settings retain the agent configuration. |

Use either `conversation_id` or `history`. The upstream API ignores earlier messages when `conversation_id` is present; this wrapper rejects a non-empty `history` combined with `conversation_id` to avoid silently dropping it.

`override_configuration` accepts these fields from the API reference:

| Settings | Fields |
| --- | --- |
| Generation | `system_prompt`, `model`, `max_new_tokens`, `top_p`, `temperature` |
| Retrieval | `top_k_retrieved_chunks`, `lexical_alpha`, `semantic_alpha`, `use_optimized_context` |
| Filtering | `enable_filter`, `filter_prompt`, `filter_model` |
| Reranking | `enable_rerank`, `reranker`, `top_k_reranked_chunks`, `rerank_instructions`, `reranker_score_filter_threshold` |

Model IDs and setting values must be supported by the selected agent. These are request overrides, not permanent changes to the agent.

## Prerequisites

- Python 3.10 or newer
- An MCP-compatible host such as Claude Desktop, Cursor, or another MCP client
- A production-issued access token and refresh token from the token portal
- Access to the token portal's HTTPS refresh endpoint
- An OS credential store supported by Python `keyring` (recommended for persistence across restarts)

## 1. Create the project

```bash
mkdir qualcomm-ai-mcp-server
cd qualcomm-ai-mcp-server

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install "mcp>=1.0.0,<2" "httpx>=0.27.0,<1" "pydantic>=2.11,<3" "keyring>=25.0,<26"
```

Activate the virtual environment using the command for your shell. For example:

```powershell
.venv\Scripts\Activate.ps1
```

For Bash or Zsh, use `source .venv/bin/activate` instead.

## 2. Configure the bearer token

Set the token in the shell where the MCP server will run. Copy the access token from the Qualcomm AI Token Portal, and do not commit it to Git.

```bash
export QUALCOMM_AI_API_TOKEN="paste-your-production-issued-token-here"
export QUALCOMM_AI_REFRESH_TOKEN="paste-your-refresh-token-here"
export QUALCOMM_AI_TOKEN_REFRESH_URL="https://<your-portal-host>/api/mcp/refresh"
export QUALCOMM_AI_WORKSPACE="dragon"
export QUALCOMM_AI_AGENT_ID="d6c12230-018a-4b99-af1b-e0bed13dd49a"
```

The workspace and agent variables are optional: the defaults are `dragon` and the agent UUID shown above. A tool call can override the agent with `agent_id`; its access is still governed by the bearer token and workspace. Set `QUALCOMM_AI_TOKEN_REFRESH_URL` to the deployed portal's `/api/mcp/refresh` endpoint. For local development, use `http://localhost:3000/api/mcp/refresh`. The tokenized download fills the refresh URL from the current portal origin and adds the current access token's expiry automatically, enabling proactive refresh. If the expiry is omitted during manual setup, the server refreshes after a gateway `401` instead.

The MCP server stores a rotated refresh token in the operating system credential store under the `qualcomm-ai-mcp` service. If a keychain is unavailable, it keeps the token only in memory and requires the environment value again after restart; it never writes the token to a plaintext file. Do not put the Auth0 client secret in the MCP environment.

## 3. Create the MCP server and tool

Create a file named `server.py`:

```python
from __future__ import annotations

import os
import asyncio
import logging
import time
from datetime import datetime, timezone
from typing import Annotated, Any, Literal
from uuid import UUID

import httpx
import keyring
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, ConfigDict, Field


APIGEE_AGENTS_URL = (
    "https://dev.apigwx-op.qualcomm.com/ce/v1/contextual-ai-harness/agents"
)
DEFAULT_AGENT_ID = "d6c12230-018a-4b99-af1b-e0bed13dd49a"
TOKEN_REFRESH_URL = os.getenv("QUALCOMM_AI_TOKEN_REFRESH_URL", "").strip()
REFRESH_BUFFER_SECONDS = 60
KEYRING_SERVICE = "qualcomm-ai-mcp"
KEYRING_ACCOUNT = os.getenv("QUALCOMM_AI_KEYRING_ACCOUNT", "default")

logger = logging.getLogger("qualcomm-ai-agent")

mcp = FastMCP("qualcomm-ai-agent")


class QueryModel(BaseModel):
    model_config = ConfigDict(extra="forbid")


class ConversationMessage(QueryModel):
    role: Literal["user", "assistant", "system", "knowledge"]
    content: str
    custom_tags: list[str] | None = None


FilterValue = str | int | float | bool


class DocumentFilter(QueryModel):
    field: str = Field(description="Document metadata field to match.")
    operator: Literal[
        "equals", "containsany", "exists", "startswith", "gt", "gte",
        "lt", "lte", "notequals", "between", "wildcard",
    ]
    value: FilterValue | list[FilterValue] | None = Field(
        default=None, description="Comparison value; omit for the exists operator."
    )


class CompositeFilter(QueryModel):
    operator: Literal["AND", "OR", "AND_NOT"]
    filters: list[DocumentFilter | CompositeFilter]


class StructuredOutput(QueryModel):
    type: Literal["JSON"] = "JSON"
    json_schema: dict[str, Any] = Field(description="JSON Schema for the generated answer.")


class QueryOverrides(QueryModel):
    # These names match AgentEvaluationConfigOverride in the Query API reference.
    system_prompt: str | None = None
    filter_prompt: str | None = None
    model: str | None = None
    max_new_tokens: int | None = None
    top_p: float | None = None
    temperature: float | None = None
    top_k_retrieved_chunks: int | None = None
    top_k_reranked_chunks: int | None = None
    enable_filter: bool | None = None
    filter_model: str | None = None
    enable_rerank: bool | None = None
    reranker: str | None = None
    lexical_alpha: float | None = None
    semantic_alpha: float | None = None
    rerank_instructions: str | None = None
    reranker_score_filter_threshold: float | None = None
    use_optimized_context: bool | None = None


def load_refresh_token() -> str | None:
    try:
        stored = keyring.get_password(KEYRING_SERVICE, KEYRING_ACCOUNT)
    except Exception:
        logger.warning("OS credential store unavailable; refresh token will remain in memory only.")
        stored = None
    if stored and stored.strip():
        return stored.strip()
    configured = os.getenv("QUALCOMM_AI_REFRESH_TOKEN", "").strip()
    return configured or None


def save_refresh_token(token: str) -> None:
    try:
        keyring.set_password(KEYRING_SERVICE, KEYRING_ACCOUNT, token)
    except Exception:
        logger.warning("OS credential store unavailable; rotated refresh token will not persist after restart.")


def parse_expiry(value: str | None) -> float | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
    except ValueError:
        return None


def format_expiry(timestamp: float) -> str:
    return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat().replace("+00:00", "Z")


class TokenState:
    def __init__(self) -> None:
        self.access_token = os.getenv("QUALCOMM_AI_API_TOKEN", "").strip() or None
        self.refresh_token = load_refresh_token()
        self.expires_at = parse_expiry(os.getenv("QUALCOMM_AI_API_TOKEN_EXPIRES_AT"))
        self.lock = asyncio.Lock()


token_state = TokenState()


async def refresh_access_token_locked() -> dict[str, Any]:
    if not token_state.refresh_token:
        raise RuntimeError("QUALCOMM_AI_REFRESH_TOKEN is not configured.")
    if not TOKEN_REFRESH_URL:
        raise RuntimeError("QUALCOMM_AI_TOKEN_REFRESH_URL is not configured.")

    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.post(
            TOKEN_REFRESH_URL,
            headers={"Accept": "application/json"},
            json={"refresh_token": token_state.refresh_token},
        )

    if response.is_error:
        raise RuntimeError(f"Token refresh failed with HTTP {response.status_code}.")
    try:
        result = response.json()
    except ValueError as exc:
        raise RuntimeError("Token refresh returned a non-JSON response.") from exc
    if not isinstance(result, dict):
        raise RuntimeError("Token refresh returned an invalid response.")

    access_token = result.get("access_token")
    expires_in = result.get("expires_in")
    if not isinstance(access_token, str) or not access_token.strip() or not isinstance(expires_in, (int, float)) or expires_in <= 0:
        raise RuntimeError("Token refresh returned an invalid token response.")

    rotated_refresh_token = result.get("refresh_token")
    token_state.access_token = access_token
    token_state.expires_at = time.time() + float(expires_in)
    if isinstance(rotated_refresh_token, str) and rotated_refresh_token.strip():
        token_state.refresh_token = rotated_refresh_token.strip()
        save_refresh_token(token_state.refresh_token)

    return {
        "expires_at": format_expiry(token_state.expires_at),
        "expires_in": int(expires_in),
        "refresh_token_rotated": isinstance(rotated_refresh_token, str) and bool(rotated_refresh_token.strip()),
    }


async def get_access_token() -> str:
    async with token_state.lock:
        if not token_state.access_token:
            raise RuntimeError("QUALCOMM_AI_API_TOKEN is not configured.")
        if token_state.expires_at is not None and token_state.expires_at <= time.time() + REFRESH_BUFFER_SECONDS:
            await refresh_access_token_locked()
        return token_state.access_token


@mcp.tool()
async def refresh_contextual_token() -> dict[str, Any]:
    """Refresh the API access token through the token portal.

    The refresh token is never sent to the API gateway or returned by this tool.
    """
    async with token_state.lock:
        metadata = await refresh_access_token_locked()
    return {"status": "refreshed", **metadata}


async def send_query(
    access_token: str,
    selected_agent: UUID,
    params: dict[str, str],
    payload: dict[str, Any],
) -> httpx.Response:
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Contextual-Workspace": os.getenv("QUALCOMM_AI_WORKSPACE", "dragon"),
        "Content-Type": "application/json",
        "Accept": "application/json",
    }
    async with httpx.AsyncClient(timeout=120.0) as client:
        return await client.post(
            f"{APIGEE_AGENTS_URL}/{selected_agent}/query/acl",
            headers=headers,
            params=params,
            json=payload,
        )


@mcp.tool()
async def query_contextual_agent(
    message: Annotated[str, Field(description="Latest user question. Required.")],
    agent_id: Annotated[UUID | None, Field(description="Agent UUID; defaults to QUALCOMM_AI_AGENT_ID or the supplied agent.")] = None,
    conversation_id: Annotated[UUID | None, Field(description="Reuse the conversation_id returned by a previous query. Do not combine with history.")] = None,
    history: Annotated[list[ConversationMessage] | None, Field(description="Earlier messages in chronological order. The tool appends message as the latest user turn.")] = None,
    custom_tags: Annotated[list[str] | None, Field(description="Tags attached to the latest user message.")] = None,
    retrievals_only: Annotated[bool, Field(description="Return retrieved content and metadata without generating an answer.")] = False,
    include_retrieval_content_text: Annotated[bool, Field(description="Include retrieved source text alongside the answer. Retrieval-only mode includes it automatically.")] = False,
    llm_model_id: Annotated[str | None, Field(description="Optional fine-tuned or aligned LLM model ID available to this agent.")] = None,
    structured_output: Annotated[StructuredOutput | None, Field(description="Request an answer matching a JSON Schema; type must be JSON.")] = None,
    documents_filters: Annotated[DocumentFilter | CompositeFilter | None, Field(description="Filter document metadata, including nested AND/OR/AND_NOT groups. Use lowercase values or field.keyword for non-equals comparisons, per the API reference.")] = None,
    override_configuration: Annotated[QueryOverrides | None, Field(description="Per-query generation, retrieval, reranking, and prompt overrides. Omitted fields retain agent settings.")] = None,
) -> dict[str, Any]:
    """Query a Qualcomm AI agent through the gateway.

    Send just message for a simple query. For follow-ups, provide conversation_id
    or history. Use retrievals_only for search, documents_filters to narrow sources,
    and structured_output for JSON answers. Returns the full JSON API response.
    Reference: see the local Qualcomm AI API Reference in the portal.
    """
    if not message.strip():
        raise ValueError("message must not be empty")
    if conversation_id is not None and history:
        raise ValueError("Use conversation_id or history, not both.")

    selected_agent = agent_id or UUID(os.getenv("QUALCOMM_AI_AGENT_ID", DEFAULT_AGENT_ID))
    messages = [item.model_dump(exclude_none=True) for item in (history or [])]
    latest_message: dict[str, Any] = {"role": "user", "content": message}
    if custom_tags is not None:
        latest_message["custom_tags"] = custom_tags
    messages.append(latest_message)

    # MCP returns one JSON result; upstream SSE streaming is deliberately disabled.
    payload: dict[str, Any] = {"messages": messages, "stream": False}
    if conversation_id is not None:
        payload["conversation_id"] = str(conversation_id)
    if llm_model_id is not None:
        payload["llm_model_id"] = llm_model_id
    if structured_output is not None:
        payload["structured_output"] = structured_output.model_dump(exclude_none=True)
    if documents_filters is not None:
        # Preserve explicitly supplied null/false/0 values, including nested filters.
        payload["documents_filters"] = documents_filters.model_dump(exclude_unset=True)
    if override_configuration is not None:
        payload["override_configuration"] = override_configuration.model_dump(exclude_none=True)

    # These two flags belong in the URL query string, not the JSON request body.
    params = {
        "retrievals_only": str(retrievals_only).lower(),
        "include_retrieval_content_text": str(include_retrieval_content_text).lower(),
    }
    response = await send_query(await get_access_token(), selected_agent, params, payload)
    if response.status_code == 401:
        async with token_state.lock:
            await refresh_access_token_locked()
            refreshed_access_token = token_state.access_token
        if not refreshed_access_token:
            raise RuntimeError("Token refresh did not return an access token.")
        response = await send_query(refreshed_access_token, selected_agent, params, payload)

    if response.is_error:
        raise RuntimeError(f"Qualcomm AI API returned HTTP {response.status_code}.")
    try:
        result = response.json()
    except ValueError as exc:
        raise RuntimeError("The gateway returned a non-JSON response.") from exc
    if not isinstance(result, dict):
        raise RuntimeError("Expected a JSON object from the Qualcomm AI Query API.")
    return result


if __name__ == "__main__":
    mcp.run(transport="stdio")
```

## 4. Run the server

With the virtual environment activated and the token environment configured:

```bash
python server.py
```

The process waits for MCP messages on standard input. It is normally started and managed by the MCP host instead of being run manually in a terminal. Do not add ordinary `print()` logging to stdout because stdout is reserved for MCP protocol messages; use stderr or Python's logging module if diagnostics are needed.

## 5. Register it with an MCP host

For an MCP host that supports a JSON server configuration, use an absolute path to the virtual environment's Python executable and the server file. A typical configuration looks like this:

```json
{
  "mcpServers": {
    "qualcomm-ai-agent": {
      "command": "/absolute/path/to/qualcomm-ai-mcp-server/.venv/bin/python",
      "args": ["/absolute/path/to/qualcomm-ai-mcp-server/server.py"],
      "env": {
        "QUALCOMM_AI_API_TOKEN": "paste-your-production-issued-token-here",
        "QUALCOMM_AI_REFRESH_TOKEN": "paste-your-refresh-token-here",
        "QUALCOMM_AI_TOKEN_REFRESH_URL": "https://<your-portal-host>/api/mcp/refresh",
        "QUALCOMM_AI_WORKSPACE": "dragon",
        "QUALCOMM_AI_AGENT_ID": "d6c12230-018a-4b99-af1b-e0bed13dd49a"
      }
    }
  }
}
```

Use the host's supported secret or environment-variable mechanism when available instead of putting the token directly in a checked-in configuration file. Replace the example command and script paths with absolute paths appropriate for your MCP host and operating system.

After restarting the MCP host, it should discover `query_contextual_agent` and `refresh_contextual_token`. If replacing an older version of this server, reload its tools in the host so it picks up the refresh tool.

The MCP server refreshes automatically when the configured access token is within 60 seconds of expiry. It also refreshes once and retries when the gateway returns HTTP 401. The explicit `refresh_contextual_token` tool is useful for checking the portal refresh connection without exposing either token to the MCP host's model response.

## 6. Example tool calls

These are MCP tool arguments for `query_contextual_agent`, not raw HTTP request bodies. The MCP SDK validates them and converts nested objects into the Python models above.

### Ask a question

```json
{
  "message": "What does the onboarding policy say about account access?",
  "include_retrieval_content_text": true
}
```

### Continue a conversation

Replace the sample UUID with the `conversation_id` returned by the previous call. Keep the same agent and workspace.

```json
{
  "message": "Which exceptions does that policy allow?",
  "conversation_id": "00000000-0000-4000-8000-000000000001"
}
```

Alternatively, provide explicit history:

```json
{
  "message": "Summarize the approval process in three steps.",
  "history": [
    {"role": "user", "content": "How do users request account access?"},
    {"role": "assistant", "content": "They submit an access request for approval."}
  ],
  "custom_tags": ["onboarding"]
}
```

### Retrieve sources with metadata filters

Use metadata fields and values present in your datastore. This example combines an exact status match with a nested category-or-tag condition:

```json
{
  "message": "Find current account access policies.",
  "retrievals_only": true,
  "documents_filters": {
    "operator": "AND",
    "filters": [
      {"field": "status", "operator": "equals", "value": "active"},
      {
        "operator": "OR",
        "filters": [
          {"field": "category", "operator": "containsany", "value": ["policy", "hr"]},
          {"field": "tags", "operator": "exists"}
        ]
      }
    ]
  },
  "override_configuration": {
    "top_k_retrieved_chunks": 20,
    "top_k_reranked_chunks": 5,
    "enable_rerank": true
  }
}
```

A single filter such as `{"field": "status", "operator": "equals", "value": "active"}` also works. Supported comparison operators are `equals`, `containsany`, `exists`, `startswith`, `gt`, `gte`, `lt`, `lte`, `notequals`, `between`, and `wildcard`; logical groups use `AND`, `OR`, or `AND_NOT`. For non-`equals` comparisons, follow the API's guidance to use lowercase values and/or `field.keyword`.

### Request a structured answer and generation settings

```json
{
  "message": "Summarize the account access policy and list the key steps.",
  "structured_output": {
    "type": "JSON",
    "json_schema": {
      "type": "object",
      "properties": {
        "summary": {"type": "string"},
        "steps": {"type": "array", "items": {"type": "string"}}
      },
      "required": ["summary", "steps"],
      "additionalProperties": false
    }
  },
  "override_configuration": {
    "temperature": 0.2,
    "max_new_tokens": 1024
  }
}
```

The tool preserves the API response envelope rather than extracting just the generated text. Reuse `conversation_id` for follow-ups; inspect `message` or `outputs` for the answer and `retrieval_contents`, `attributions`, or `groundedness_scores` when returned by the agent. Retrieval-only queries may omit a generated answer. `structured_output` controls the generated answer's structure, not the entire response envelope.

## Troubleshooting

### HTTP 401 or 403

Confirm that the token is current, production-issued, and available to the process that starts `server.py`. Also verify that the token is allowed to use the `dragon` workspace and this agent.

### HTTP 422

Check the option names and value types against the Query API reference. The wrapper appends the latest user message and sends retrieval flags as URL parameters. Nested input objects reject unknown fields. If the gateway rejects an optional setting, retry with only `message` and check which options the deployed gateway and agent support.

### The MCP host cannot start the server

Use absolute paths for both `command` and `args`, confirm that the virtual environment contains `mcp`, `httpx`, and `pydantic`, and check the host's stderr or logs. The server must use stdio transport and must not write diagnostic output to stdout.

## Security notes

- Treat the bearer token like a password. Do not commit it, paste it into source code, or include it in screenshots.
- Keep the MCP server on a trusted machine and limit access to the MCP host configuration.
- Remove or rotate the token if it is exposed.
- The tool sends the supplied message, history, filters, and query settings to the configured Apigee endpoint. Apply your organization's data-handling rules before sending sensitive content.