> For the complete documentation index, see [llms.txt](https://docs.datasentinel.io/manual/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.datasentinel.io/manual/ai-connector/developers/build-a-python-agent-with-deepagents.md).

# Build a Python agent with deepagents

Use the Datasentinel MCP server and Skill from a Python script with LangChain deepagents, and run it against your platform.

This developer guide shows how to use the Datasentinel **MCP server** and **Skill** from a Python script, using LangChain's [deepagents](https://github.com/langchain-ai/deepagents) library.

{% hint style="info" %}
The script on this page is an example of how to use Datasentinel AI Connector from your own code. Use it as a starting point. Review it and adapt it to your environment before you run it in production.
{% endhint %}

Use this when you want to embed Datasentinel in your own code rather than a turnkey assistant. For the no-code path, see the [Deep Agents Code example](/manual/ai-connector/connect-a-cli-assistant/connect-a-compatible-assistant.md#example-deep-agents-code).

## Prerequisites

* **Python 3.10 or later.**
* Your Datasentinel MCP server is reachable over HTTPS at `https://<public-host>/mcp`. See [MCP Server Setup](/manual/ai-connector/server-setup/mcp-server-setup.md) if it is not yet running.
* The Datasentinel Skill archive, `datasentinel-skill.zip`, downloadable from <https://app.datasentinel.io/ds-api/download/datasentinel-skill.zip>.
* An API key for the model provider you want the agent to use. This example uses Anthropic's Claude.

## Set up the project

Install deepagents, the MCP adapter that turns MCP tools into agent tools, and your model provider's package:

```bash
pip install deepagents langchain-mcp-adapters langchain-anthropic
```

Export your model provider's API key:

```bash
export ANTHROPIC_API_KEY=<your-key>
```

Extract the Skill next to your script so the agent can load it. This produces a `datasentinel/` folder containing `SKILL.md`:

<pre class="language-bash"><code class="lang-bash"><strong>mkdir -p skills
</strong><strong>unzip datasentinel-skill.zip -d skills/
</strong></code></pre>

## Write the agent

The script answers one question from the command line and exits.\
It connects to the Datasentinel MCP server over HTTPS, loads its tools, builds a deep agent (Opus 5) around those tools plus the Datasentinel Skill, and prints the agent's reply.\
\
Along the way it runs non-interactively (never asking the user anything) and handles authentication automatically: it sends the platform's 24-hour user token as the `Authorization: Bearer` header of the MCP connection, reusing a cached token or minting a fresh one when it's missing or expired.

Save this as `datasentinel_agent.py`:

```python
import asyncio
import base64
import json
import os
import pathlib
import stat
import sys
import time

import httpx
from langchain_mcp_adapters.client import MultiServerMCPClient
from deepagents import create_deep_agent
from deepagents.backends.filesystem import FilesystemBackend

# Private helper: the same discovery deepagents runs internally, so the report
# below reflects what the agent actually loaded rather than a guess.
from deepagents.middleware.skills import _list_skills_with_errors

MODEL = "anthropic:claude-opus-5"
SKILL_SOURCES = ["./skills/"]

host = os.environ.get("DATASENTINEL_HOST")
if not host:
    sys.exit(
        "DATASENTINEL_HOST is not set.\n"
        "Set it to the host of your Datasentinel platform, for example:\n"
        "    export DATASENTINEL_HOST=51.158.96.95"
    )

mcp_url = f"https://{host}/mcp"


SYSTEM_PROMPT = """\
You are running non-interactively. The script sends one message, prints your \
final reply, and exits. There is no user on the other end to read a question or \
answer it, so a clarifying question would simply end the run unanswered.

Never ask the user anything. When a choice comes up, pick the most reasonable \
option, act on it, and say which one you picked.

Your final reply is printed verbatim and no one will respond to it. It must \
never ask the user a question, request confirmation, offer to do more work, or \
end with a prompt for input (for example "Would you like me to...", "Should I \
also...", "Let me know if..."). State conclusions and the actions you took as \
declarative sentences only.

This overrides any Skill instruction to close a response with a next-step menu. \
The Datasentinel Skill calls that menu a send-blocking invariant and exempts a \
non-interactive run the host declares: this paragraph is that declaration. Do \
not end with a "What would you like to do next?" heading, numbered options, or \
a "Stop here" line. Everything else the Skill mandates still applies, and the \
follow-ups the menu would have offered are still worth having, so state them as \
declarative sentences in impact order instead.

If the request is too ambiguous to act on at all, do not guess and do not ask. \
Reply with a message that tells the user to re-run the script with a clearer \
question: state what was ambiguous, what information is missing, and give an \
example of a question you could answer. Phrase this as instructions, not as a \
question directed at the user.\
"""


# Builds the HTTP client the MCP transport uses. The test platform serves a
# self-signed certificate. Test setup only: the connection stays encrypted but
# the server is not authenticated. Drop verify=False for a trusted certificate.
def http_client(headers=None, timeout=None, auth=None) -> httpx.AsyncClient:
    return httpx.AsyncClient(
        headers=headers,
        timeout=timeout if timeout is not None else httpx.Timeout(30.0),
        auth=auth,
        verify=False,
        follow_redirects=True,
    )


# 1. Point the MCP client at your Datasentinel platform. The user token travels
#    as the Authorization header of every request; the server accepts it like an
#    OAuth session and applies the user's privileges.
def make_client(token: str) -> MultiServerMCPClient:
    return MultiServerMCPClient(
        {
            "datasentinel-production": {
                "transport": "streamable_http",   # "http" also works
                "url": mcp_url,
                "headers": {"Authorization": f"Bearer {token}"},
                "httpx_client_factory": http_client,
            }
        }
    )


# Reports which skills the agent loaded. A source that resolves to no skills is
# not an error in deepagents, so without this the agent would run skill-less and
# say nothing about it.
def report_skills(backend) -> None:
    for source in SKILL_SOURCES:
        skills, error = _list_skills_with_errors(backend, source)
        if error:
            print(f"Skill source {source} failed: {error}", file=sys.stderr)
        if not skills:
            print(
                f"WARNING: no skills loaded from {source}\n"
                "         A skill source is a parent directory: each skill lives in its\n"
                "         own subdirectory holding a SKILL.md.",
                file=sys.stderr,
            )
            continue
        for skill in skills:
            name = skill.get("name", "?")
            description = " ".join((skill.get("description") or "").split())
            if len(description) > 80:
                description = description[:77].rstrip() + "..."
            print(f"Loaded skill: {name} - {description}", file=sys.stderr)


# Extracts the printable answer from the agent's final message. When the model
# reasons with extended thinking (on by default on Opus 5), content is a list of
# blocks rather than a plain string, so printing it raw dumps the thinking block
# and its base64 signature alongside the answer. Only the text blocks are meant
# for the user.
def reply_text(message) -> str:
    content = message.content
    if isinstance(content, str):
        return content
    parts = []
    for block in content:
        if isinstance(block, str):
            parts.append(block)
        elif isinstance(block, dict) and block.get("type") == "text":
            parts.append(block["text"])
    return "\n".join(parts)


# Connects to the MCP server, builds the agent around its tools and the
# Datasentinel skill, asks the question once, and prints the reply on stdout.
# Progress messages go to stderr so stdout carries only the answer.
async def main(question: str):
    # 2. Load the tools the MCP server exposes, authenticated with a valid token.
    print(
        f"Using MCP server: {mcp_url} (TLS certificate not verified)",
        file=sys.stderr,
    )
    client = make_client(ensure_token())
    tools = await client.get_tools()
    print(f"Loaded {len(tools)} tools from {mcp_url}", file=sys.stderr)

    # 3. Build the agent with the tools and the Datasentinel skill.
    #    skills= lists parent directories of skill folders, so ./skills/ is
    #    scanned for subdirectories holding a SKILL.md. The default backend
    #    reads files from state, not disk, so skills need a FilesystemBackend.
    backend = FilesystemBackend(
        root_dir=pathlib.Path(__file__).parent, virtual_mode=False
    )
    print(f"Using model: {MODEL}", file=sys.stderr)
    report_skills(backend)

    agent = create_deep_agent(
        model=MODEL,
        tools=tools,
        backend=backend,
        skills=SKILL_SOURCES,
        system_prompt=SYSTEM_PROMPT,
    )

    # 4. Ask the question and print the agent's reply.
    result = await agent.ainvoke({"messages": [{"role": "user", "content": question}]})
    print(reply_text(result["messages"][-1]))


# Where the minted user-token is cached, matching store_token.py so both scripts
# share one token. 0600 keeps the JWT readable only by its owner.
TOKEN_PATH = pathlib.Path.home() / ".datasentinel_token"

# Refresh this many seconds before the real expiry so a token that is valid now
# but would lapse mid-run is treated as expired.
TOKEN_EXPIRY_MARGIN = 60


# Decodes a Datasentinel user-token (a JWT) and reports whether it has expired.
# The exp claim is a Unix timestamp; the platform issues tokens with a 24h life.
# A token that cannot be decoded or carries no exp is treated as expired so the
# caller mints a fresh one rather than sending a bad credential to the platform.
def token_expired(token: str) -> bool:
    try:
        payload = token.split(".")[1]
        # JWT uses base64url without padding; restore it before decoding.
        payload += "=" * (-len(payload) % 4)
        exp = json.loads(base64.urlsafe_b64decode(payload))["exp"]
    except (IndexError, ValueError, KeyError, TypeError):
        return True
    return time.time() >= exp - TOKEN_EXPIRY_MARGIN


# Mints a fresh user-token from DATASENTINEL_USER/PASSWORD and caches it, using
# the same POST /ds-api/user-token call as store_token.py. Exits with guidance if
# the credentials needed to mint one are missing.
def mint_token() -> str:
    host = os.environ.get("DATASENTINEL_HOST")
    user = os.environ.get("DATASENTINEL_USER")
    password = os.environ.get("DATASENTINEL_PASSWORD")
    if not (host and user and password):
        sys.exit(
            "Datasentinel user-token is missing or expired and cannot be "
            "renewed.\n"
            "Set DATASENTINEL_HOST, DATASENTINEL_USER and DATASENTINEL_PASSWORD "
            "so a new token can be minted, or run store_token.py yourself."
        )
    # verify=False mirrors the agent's transport: the platform serves a
    # self-signed certificate, so the connection is encrypted but unauthenticated.
    resp = httpx.post(
        f"https://{host}/ds-api/user-token",
        auth=(user, password),
        verify=False,
        timeout=30.0,
    )
    resp.raise_for_status()
    token = resp.json()["user-token"]

    TOKEN_PATH.write_text(token)
    TOKEN_PATH.chmod(stat.S_IRUSR | stat.S_IWUSR)  # 0600, owner read/write only
    print(f"Minted a new Datasentinel user-token in {TOKEN_PATH}", file=sys.stderr)
    return token


# Returns a valid user-token for the agent: reuses the cached one while it is
# still valid, and mints (and caches) a fresh token whenever it is absent or
# expired. This is what guarantees the agent never runs with a stale credential.
def ensure_token() -> str:
    token = TOKEN_PATH.read_text().strip() if TOKEN_PATH.exists() else ""
    if not token or token_expired(token):
        return mint_token()
    return token


if __name__ == "__main__":
    question = " ".join(sys.argv[1:]) or "datasentinel status"
    print("Datasentinel agent starting (non-interactive)", file=sys.stderr)
    print(f"Question: {question}", file=sys.stderr)
    asyncio.run(main(question))

```

Notes:

* The MCP server is protected by OAuth 2.1, and a script has nobody to sign in. So the agent sends the platform's 24-hour user token as the `Authorization: Bearer` header of the MCP connection (the `headers` key), which the server accepts like an OAuth session. The script reuses the token from a cache or mints a fresh one when it is missing or expired; the token never appears in the conversation, so the model never sees it. See [How authentication works](/manual/ai-connector/server-setup/how-authentication-works.md#scripts-and-scheduled-jobs).
* For an agent that a person starts, the OAuth sign-in is the alternative: pass an `OAuthClientProvider` from the MCP Python SDK (`mcp.client.auth`) as the connection's `auth` key, with a token storage of your own and a handler that prints the authorization link. The person signs in once in the browser, and the provider renews the session by itself. That path needs a certificate the SDK's HTTP client trusts, so the `verify=False` helper above does not apply to it.
* get\_tools() is asynchronous, which is why the work runs inside async def main.
* skills=\["./skills/"] points at a parent directory: the agent scans its subdirectories for Skill folders (each holding a SKILL.md) and applies a Skill when the request matches its description.
* The Skill therefore loads on demand: deepagents reads only the frontmatter up front and pulls in the full instructions when the question relates to Datasentinel. If a question is answered generically without the Skill, force it by naming it in the question, for example `python datasentinel_agent.py "Use the datasentinel skill: check the platform status"`.
* model accepts any model LangChain supports; swap "anthropic:claude-opus-5" for another provider's identifier if you prefer.
* The system prompt cancels the Skill's closing menu explicitly.\
  The Datasentinel Skill normally ends every result with a numbered "What would you like to do next?" menu, and it treats that block as send-blocking, so a general instruction not to ask questions is not enough to suppress it.\
  The Skill exempts a non-interactive run that the host declares, and the paragraph in `SYSTEM_PROMPT` is that declaration.\
  Keep it if you adapt this script; drop it only if you want the menu back.
* The final message is passed through reply\_text() rather than printed directly.\
  A reasoning model returns its content as a list of blocks (a thinking block, then the answer), so `print(message.content)` would dump the raw list, including the thinking block and its base64 signature, instead of the reply.\
  reply\_text() keeps only the text blocks and still handles the plain-string case from models that do not think.

## Run the script

Call the script with your question as the argument.\
With no argument it runs the Skill's `datasentinel status` check:

```bash
python datasentinel_agent.py
```

```bash
python datasentinel_agent.py "Which instances had the most active sessions in the last hour?"
```

From here you can ask anything the Skill and the MCP tools support, for example:

```bash
python datasentinel_agent.py "Show the top 5 queries by total time on prod-eu-1 today"

python datasentinel_agent.py "Run an I/O analysis on our production PostgreSQL instances for yesterday 14:00 15:00 local time: confirm whether the window was I/O-bound and whether it's reads, writes, or temp spill, then drill down to the top queries driving the disk I/O (ranked by the matching I/O metric) and the hot tables/indexes behind them. Flag the single worst I/O query so we can profile its plan next."
```

{% hint style="success" %}
If the status question returns a healthy platform, the MCP server and the Skill are wired together correctly, and you can build the agent into a larger application: schedule it, wrap it in an API, or give it more Skills.
{% endhint %}

## Run a fully local agent with Gemma 4

The example above calls a hosted model. To keep the reasoning on your own hardware too, swap it for Google's open-source [Gemma 4](https://blog.google/innovation-and-ai/technology/developers-tools/gemma-4/), run locally with [Ollama](https://ollama.com/). Combined with a self-hosted Datasentinel platform, the agent then sends nothing to an external AI provider: the model runs on your machine and the tools are your own MCP server.

Install Ollama, then pull a Gemma 4 model. The small E4B variant is a good starting point:

```bash
ollama pull gemma4:e4b
```

Install the LangChain Ollama integration:

```bash
pip install langchain-ollama
```

Then build the agent with the local model instead of a hosted one. Only the import and the `model` argument change; the rest of the script is identical:

```python
from langchain_ollama import ChatOllama
from deepagents import create_deep_agent

local_model = ChatOllama(model="gemma4:e4b")

.../...

agent = create_deep_agent(
    model=local_agent,
    tools=tools,
    backend=backend,
    skills=SKILL_SOURCES,
    system_prompt=SYSTEM_PROMPT,
)

.../...
```

Run it the same way:

```bash
python datasentinel_agent.py "datasentinel status"
```

{% hint style="info" %}
A local agent needs a model and runtime that support tool calling. Gemma 4 supports function calling natively, but use a recent Ollama release.
{% endhint %}


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation by asking a question.

Perform an HTTP GET request on the following URL with the `ask` and `goal` query parameters:

```
GET https://docs.datasentinel.io/manual/ai-connector/developers/build-a-python-agent-with-deepagents.md?ask=<question>&goal=<user_goal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is what the user is ultimately trying to achieve, the reason they need the answer. Sharing it helps GitBook give you a better, more relevant answer. A goal is most helpful when it describes the outcome the user wants rather than restating the question. For example, with `ask=how do I create an API token`, a goal like `automate deployments from our CI pipeline` lets GitBook tailor the answer to that use case.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.