> ## Documentation Index
> Fetch the complete documentation index at: https://docs.blaxel.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Develop agents in Python
> Build, test, and run a custom Python agent locally with the Blaxel SDK and CLI, with code examples for LangChain, CrewAI, and other frameworks.
You can bring your **custom agents developed in Python** and deploy them to Blaxel with our developer tools ([Blaxel CLI](../cli-reference/introduction), [GitHub app](/Agents/Github-integration), GitHub action, etc.). You can develop agents using frameworks like LangChain, Google ADK, OpenAI Agents SDK; or your own custom code.
## Quickstart
It is required to [have *uv* installed](https://docs.astral.sh/uv/getting-started/installation/) to use the following command.
You can quickly **initialize a new project from scratch** by using CLI command `bl new`.
```bash theme={null}
bl new agent
```
This will create a pre-scaffolded local repo where your entire code can be added. You can choose the base agentic framework for the template.
In the generated folder, you'll find a standard server in the entrypoint file `main.py`. While you typically won't need to modify this file, you can add specific logic there if needed. Your main work will focus on the `agent.py` file. Blaxel's development paradigm lets you leverage its hosting capabilities without modifying your agent's core logic.
### Requirements & limitations
Agents Hosting have few requirements or limitations:
* The only requirement to deploy an app on Agents Hosting is that it exposes an HTTP API server which is bound on `HOST` (for the host) and `PORT` (for the port). **These two environment variables are required for the host+port combo.**
* You can use [express](https://expressjs.com/), [fastify](https://fastify.dev/), [FastAPI](https://fastapi.tiangolo.com/), etc. for this.
* Agents deployed on Blaxel Agents Hosting have a maximum runtime of 15 minutes. This limit does not apply to [Sandboxes](../Sandboxes/Overview) or [Batch Jobs](../Jobs/Overview), which have their own runtime limits.
* The synchronous endpoint closes the connection after **100 seconds** if no data flows through the API. If your agent streams back responses, the connection resets with each chunk streamed. For example, if your agent processes a request for 5 minutes while streaming data, the connection stays open. However, if it goes 100 seconds without sending any data — even while calling external APIs — the connection will close.
## Accessing resources with Blaxel SDK
[Blaxel SDK](../sdk-reference/introduction) provides methods to programmatically access and integrate various resources hosted on Blaxel into your agent's code, such as: [model APIs](../Models/Overview), [tool servers](../Functions/Overview), [sandboxes](../Sandboxes/Overview), [batch jobs](../Jobs/Overview), or [other agents](Overview). The SDK handles authentication, secure connection management and telemetry automatically.
### Connect to a model API
Blaxel SDK provides a helper to connect to a [model API](../Models/Overview) defined on Blaxel from your code. This allows you to avoid managing a connection with the model API by yourself. Credentials remain stored securely on Blaxel.
```python theme={null}
from blaxel.{FRAMEWORK_NAME} import bl_model
model = await bl_model("Model-name-on-Blaxel");
```
The model is automatically converted to your chosen framework's format based on the `FRAMEWORK_NAME` specified in the import.
Available frameworks :
* [LangGraph/LangChain](https://python.langchain.com/docs/concepts/chat_models/) : `langgraph`
* [CrewAI](https://docs.crewai.com/concepts/llms) : `crewai`
* [LlamaIndex](https://docs.llamaindex.ai/en/stable/module_guides/models/llms/) : `llamaindex()`
* [OpenAI Agents](https://github.com/openai/openai-agents-python): `openai()`
* [Pydantic AI Agents](https://github.com/pydantic/pydantic-ai): `pydantic()`
* [Google ADK](https://github.com/google/adk-python/blob/main/src/google/adk/models/lite_llm.py): `googleadk()`
For example, to connect to model `my-model` in a *LlamaIndex* agent:
```python theme={null}
from blaxel.llamaindex import bl_model
model = await bl_model("my-model")
```
### Connect to tools
Blaxel SDK provides a helper to connect to [pre-built or custom tool servers (MCP servers)](../Functions/Overview) hosted on Blaxel from your code. This allows you to avoid managing a connection with the server by yourself. Credentials remain stored securely on Blaxel. The following method retrieves all the tools discoverable in the tool server.
```python theme={null}
from blaxel.{FRAMEWORK_NAME} import bl_tools
await bl_tools(['Tool-Server-name-on-Blaxel'])
```
Like for a model, the retrieved tools are automatically converted to the format of the framework you want to use based on the Blaxel SDK package imported. Available frameworks are `langgraph` ([LangGraph/Langchain](https://python.langchain.com/api_reference/core/tools/langchain_core.tools.structured.StructuredTool.html)), `llamaindex` ([LlamaIndex](https://docs.llamaindex.ai/en/stable/module_guides/deploying/agents/tools/)), `crewai` ([CrewAI](https://docs.crewai.com/concepts/tools)), `openai` ([OpenAI Agents](https://github.com/openai/openai-agents-python)), `pydantic` ([PydanticAI Agents](https://github.com/pydantic/pydantic-ai)) and `googleadk` ([Google ADK](https://github.com/google/adk-python/blob/main/src/google/adk/tools/base_tool.py)).
You can develop agents by **mixing tools defined locally in your agents, and tools defined as remote servers**. Using separated tools prevents monolithic designs which make maintenance easier in the long run. Let's look at a practical example combining remote and local tools. The code below uses two tools:
1. `blaxel-search`: A remote tool server on Blaxel providing web search functionality (learn how to create your own MCP servers [here](../Functions/Create-MCP-server))
2. `weather`: A local tool that accepts a city parameter and returns a mock weather response (always "sunny")
```python agent.py (LangGraph) theme={null}
from typing import AsyncGenerator
from blaxel.langgraph import bl_model, bl_tools
from langchain.tools import tool
from langchain_core.messages import AIMessageChunk
from langgraph.prebuilt import create_react_agent
@tool
def weather(city: str) -> str:
"""Get the weather in a given city"""
return f"The weather in {city} is sunny"
async def agent(input: str) -> AsyncGenerator[str, None]:
prompt = "You are a helpful assistant that can answer questions and help with tasks."
### Load tools dynamically from Blaxel, and adding a tool defined locally:
tools = await bl_tools(["blaxel-search"]) + [weather]
### Load model API dynamically from Blaxel:
model = await bl_model("gpt-4o-mini")
agent = create_react_agent(model=model, tools=tools, prompt=prompt)
messages = {"messages": [("user", input)]}
async for chunk in agent.astream(messages, stream_mode=["updates", "messages"]):
type_, stream_chunk = chunk
# This is to stream the response from the agent, filtering response from tools
if type_ == "messages" and len(stream_chunk) > 0 and isinstance(stream_chunk[0], AIMessageChunk):
msg = stream_chunk[0]
if msg.content:
if not msg.tool_calls:
yield msg.content
# This to show a call has been made to a tool, useful if you want to show the tool call in your interface
if type_ == "updates":
if "tools" in stream_chunk:
for msg in stream_chunk["tools"]["messages"]:
yield f"Tool call: {msg.name}\n"
```
```python agent.py (LlamaIndex) theme={null}
from typing import AsyncGenerator
from blaxel.llamaindex import bl_model, bl_tools
from llama_index.core.agent.workflow import AgentStream, ReActAgent
from llama_index.core.tools import FunctionTool
async def weather(city: str) -> str:
"""Get the weather in a given city"""
return f"The weather in {city} is sunny"
async def agent(input: str) -> AsyncGenerator[str, None]:
prompt = "You are a helpful assistant that can answer questions and help with tasks."
### Load tools dynamically from Blaxel, and adding a tool defined locally:
tools = await bl_tools(["blaxel-search"]) + [FunctionTool.from_defaults(async_fn=weather)]
### Load model API dynamically from Blaxel:
model = await bl_model("gpt-4o-mini")
agent = ReActAgent(llm=model, tools=tools, system_prompt=prompt)
async for event in agent.run(input).stream_events():
if isinstance(event, AgentStream):
yield event.delta
```
```python agent.py (CrewAI) theme={null}
# We have to apply nest_asyncio because crewai is not compatible with async
import nest_asyncio
nest_asyncio.apply()
from typing import AsyncGenerator
from blaxel.crewai import bl_model, bl_tools
from crewai import Agent, Crew, Task
from crewai.tools import tool
@tool("Weather")
def weather(city: str) -> str:
"""Get the weather in a given city"""
return f"The weather in {city} is sunny"
async def agent(input: str) -> AsyncGenerator[str, None]:
### Load tools dynamically from Blaxel, and adding a tool defined locally:
tools = await bl_tools(["blaxel-search"]) + [weather]
### Load model API dynamically from Blaxel:
model = await bl_model("gpt-4o-mini")
agent = Agent(
role="Weather Researcher",
goal="Find the weather in a city",
backstory="You are an experienced weather researcher with attention to detail",
llm=model,
tools=tools,
verbose=True,
)
crew = Crew(
agents=[agent],
tasks=[Task(description="Find weather", expected_output=input, agent=agent)],
verbose=True,
)
result = crew.kickoff()
yield result.raw
```
```python agent.py (OpenAI Agents) theme={null}
from typing import AsyncGenerator
from agents import Agent, RawResponsesStreamEvent, Runner, function_tool
from blaxel.openai import bl_model, bl_tools
from openai.types.responses import ResponseTextDeltaEvent
@function_tool()
async def weather(city: str) -> str:
"""Get the weather in a given city"""
return f"The weather in {city} is sunny"
async def agent(input: str) -> AsyncGenerator[str, None]:
### Load tools dynamically from Blaxel, and adding a tool defined locally:
tools = await bl_tools(["blaxel-search"]) + [weather]
### Load model API dynamically from Blaxel:
model = await bl_model("gpt-4o-mini")
agent = Agent(
name="blaxel-agent",
model=model,
tools=tools,
instructions="You are a helpful assistant.",
)
result = Runner.run_streamed(agent, input)
async for event in result.stream_events():
if isinstance(event, RawResponsesStreamEvent) and isinstance(event.data, ResponseTextDeltaEvent):
yield event.data.delta
```
```python agent.py (Pydantic AI Agents) theme={null}
from typing import AsyncGenerator
from blaxel.pydantic import bl_model, bl_tools
from pydantic_ai import Agent, CallToolsNode, Tool
from pydantic_ai.messages import ToolCallPart
from pydantic_ai.models import ModelSettings
def weather(city: str) -> str:
"""Get the weather in a given city"""
return f"The weather in {city} is sunny"
async def agent(input: str) -> AsyncGenerator[str, None]:
prompt = "You are a helpful assistant that can answer questions and help with tasks."
### Load tools dynamically from Blaxel, and adding a tool defined locally:
tools = await bl_tools(["blaxel-search"]).to_pydantic() + [Tool(weather)]
### Load model API dynamically from Blaxel:
model = await bl_model("gpt-4o-mini").to_pydantic()
agent = Agent(model=model, tools=tools, model_settings=ModelSettings(temperature=0), system_prompt=prompt)
async with agent.iter(input) as agent_run:
async for node in agent_run:
if isinstance(node, CallToolsNode):
for part in node.model_response.parts:
if isinstance(part, ToolCallPart):
yield(f"Tool call: {part.tool_name}\n")
else:
yield part.content + "\n"
```
```python agent.py (Google ADK) theme={null}
from logging import getLogger
from typing import AsyncGenerator
from blaxel.googleadk import bl_model, bl_tools
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
logger = getLogger(__name__)
# @title Define the get_weather Tool
def get_weather(city: str) -> dict:
"""Get the weather in a given city"""
return f"The weather in {city} is sunny"
APP_NAME = "research_assistant"
session_service = InMemorySessionService()
async def agent(input: str, user_id: str = "default", session_id: str = "default") -> AsyncGenerator[str, None]:
description = "You are a helpful assistant that can answer questions and help with tasks."
prompt = """
You are a helpful weather assistant. Your primary goal is to provide current weather reports. "
When the user asks for the weather in a specific city,
You can also use a research tool to find more information about anything.
Analyze the tool's response: if the status is 'error', inform the user politely about the error message.
If the status is 'success', present the weather 'report' clearly and concisely to the user.
Only use the tool when a city is mentioned for a weather request.
"""
### Load tools dynamically from Blaxel, and adding a tool defined locally:
tools = await bl_tools(["blaxel-search"]).to_google_adk() + [get_weather]
### Load model API dynamically from Blaxel:
model = await bl_model("sandbox-openai").to_google_adk()
agent = Agent(model=model, name=APP_NAME, description=description, instruction=prompt, tools=tools)
# Create the specific session where the conversation will happen
if not session_service.get_session(app_name=APP_NAME, user_id=user_id, session_id=session_id):
session_service.create_session(
app_name=APP_NAME,
user_id=user_id,
session_id=session_id
)
logger.info(f"Session created: App='{APP_NAME}', User='{user_id}', Session='{session_id}'")
runner = Runner(
agent=agent,
app_name=APP_NAME,
session_service=session_service,
)
logger.info(f"Runner created for agent '{runner.agent.name}'.")
content = types.Content(role="user", parts=[types.Part(text=input)])
async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=content):
# Key Concept: is_final_response() marks the concluding message for the turn.
if event.is_final_response():
if event.content and event.content.parts:
# Assuming text response in the first part
yield event.content.parts[0].text
elif event.actions and event.actions.escalate: # Handle potential errors/escalations
yield f"Agent escalated: {event.error_message or 'No specific message.'}"
```
### Connect to another agent (multi-agent chaining)
Rather than using a "quick and dirty" approach where you would combine all your agents and capabilities into a single deployment, Blaxel provides a structured development paradigm based on two key principles:
* Agents can grow significantly in complexity. Monolithic architectures make long-term maintenance difficult.
* Individual agents should be reusable across multiple projects.
Blaxel supports a microservice architecture for handoffs, allowing you to call one agent from another using `bl_agent().run()` rather than combining all functionality into a single codebase.
```bash theme={null}
from blaxel.core.agents import bl_agent
first_agent_response = await bl_agent("first_agent").run(input);
second_agent_response = await bl_agent("second_agent").run(first_agent_response);
```
## Customize the agent deployment
You can set custom parameters for an agent deployment (e.g. specify the agent name, etc.) in the [`blaxel.toml` file](/deployment-reference) at the root of your directory.
Read the file structure section down below for more details.
## Instrumentation
Instrumentation happens automatically when workloads run on Blaxel. To enable telemetry, simply import the SDK in your project's entry point.
```bash theme={null}
import blaxel.core
```
When agents and tools are deployed on Blaxel, request logging and tracing happens automatically.
You do not need to call `logging.basicConfig()` in your agent code. Importing any `blaxel.*` module automatically initializes the logger with the correct formatter for the Blaxel platform.
To add your own custom logs that you can view in the Blaxel Console, use the Python default logger.
```python theme={null}
import logging
logger = logging.getLogger(__name__)
logger.info("Hello, world!")
```
## Template directory reference
### Overview
```bash theme={null}
pyproject.toml # Mandatory. This file is the standard pyproject.toml file, it defines dependencies.
blaxel.toml # This file lists configurations dedicated to Blaxel to customize the deployment. It is not mandatory.
.blaxel # This folder allows you to define custom resources using the Blaxel API specifications. These resources will be deployed along with your agent.
├── blaxel-search.yaml # Here, blaxel-search is a sandbox Web search tool we provide so you can develop your first agent. It has a low rate limit, so we recommend you use a dedicated MCP server for production.
src/
└── main.py # This file is the standard entrypoint of the project. It is used to start the server and create an endpoint bound with agent.py file.
├── agent.py # This file is the main file of your agent. It is loaded from main.py. In the template, all the agent logic is implemented here.
```
### blaxel.toml
The agent deployment can be configured via the `blaxel.toml` file in your agent directory. This file is not mandatory; if the file is not found or a required option is not set, you will be prompted for the information during deployment.
```toml theme={null}
name = "my-agent"
workspace = "my-workspace"
type = "agent"
public = false
agents = []
functions = ["blaxel-search"]
models = ["gpt-4o-mini"]
[env]
DEFAULT_CITY = "San Francisco"
[runtime]
memory = 1024
[[triggers]]
id = "trigger-async-my-agent"
type = "http-async"
[triggers.configuration]
path = "agents/my-agent/async" # This will create this endpoint on the following base URL: https://run.blaxel.ai/{YOUR-WORKSPACE}
retry = 1
[[triggers]]
id = "trigger-my-agent"
type = "http"
[triggers.configuration]
path = "agents/my-agent/sync"
retry = 1
```
* `name`, `workspace`, and `type` fields are optional and serve as default values. Any bl command run in the folder will use these defaults rather than prompting you for input.
* `agents`, `functions`, and `models` fields are also optional. They specify which resources to deploy with the agent. These resources are preloaded during build, eliminating runtime dependencies on the Blaxel control plane and dramatically improving performance.
* `public` field specifies if the agent is publicly accessible (defaults to `false`).
* `[env]` section defines environment variables that the agent can access via the SDK. Note that these are NOT [secrets](/Agents/Variables-and-secrets).
* `[runtime]` section allows to override agent deployment parameters: memory (in MB) to allocate.
* `[[triggers]]` and `[triggers.configuration]` sections define ways to send requests to the agent. You can create both [synchronous and asynchronous](/Agents/Query-agents) trigger endpoints (respectively `type = "http"` or `type = "http-async"`). A private synchronous HTTP endpoint is always created by default, even if you don’t define any trigger here.
Additionally, you can define an `[entrypoint]` section to specify how Blaxel is going to start your server:
```toml theme={null}
...
[entrypoint]
prod = "python src/main.py"
dev = "fastapi dev"
...
```
* `prod`: this is the command that will be used to serve your agent
```bash theme={null}
python src/main.py
```
* `dev`: same as prod in dev mode, it will be used with the command `--hotreload`. Example:
```bash theme={null}
fastapi dev
```
This `entrypoint` section is optional. If not specified, Blaxel will automatically detect in the agent’s content and configure your agent startup settings.
## Troubleshooting
### Wrong port or host
```text theme={null}
Default STARTUP TCP probe failed 1 time consecutively for container "agent" on port 80. The instance was not started.
Connection failed with status DEADLINE_EXCEEDED.
```
If you encounter this error when deploying your agent on Blaxel, ensure that your agent properly exposes an API server that binds to a host and port with the **required** environment variables: `HOST` & `PORT`. Blaxel automatically injects these variables during deployment.
See an example of building and deploying an agent on Blaxel with Claude Agent SDK.
Learn how to deploy your custom AI agents on Blaxel as a serverless endpoint.