railtownai 2.0.14

Railtown AI Python SDK for building AI-powered applications with enterprise integrations.

Railtown AI Python SDK

The Railtown AI Python SDK provides error tracking for applications, agent observability, and enterprise connectors for seamless integration for your agents.

Logging

1. Sign up for Railtown AI - Conductr
2. Create a project, navigate to the Project Configuration page, and copy your API key
3. In your app:
   1. Install the Railtown AI SDK: pip install railtownai
   2. Initialize Railtown AI with your API key: railtownai.init('YOUR_RAILTOWN_API_KEY')
   3. Use Python's native logging - all logs will automatically be sent to Railtown AI

Logging Basic Usage

import logging
import railtownai

# Initialize Railtown AI
railtownai.init('YOUR_RAILTOWN_API_KEY')

# Use Python's native logging - all logs are sent to Railtown AI
logging.info("User logged in", extra={"user_id": 123, "action": "login"})
logging.warning("High memory usage detected", extra={"memory_usage": "85%"})
logging.error("Database connection failed", extra={"db_host": "localhost"})

# Log exceptions with full stack traces
try:
    result = 1 / 0
except Exception:
    logging.exception("Division by zero error")

Logging with Breadcrumbs

Railtown AI supports breadcrumbs - contextual information that gets attached to log events. This is useful for tracking user actions or system state leading up to an error.

import logging
import railtownai

railtownai.init('YOUR_RAILTOWN_API_KEY')

# Add breadcrumbs throughout your application
railtownai.add_breadcrumb("User clicked login button", category="ui")
railtownai.add_breadcrumb("Validating user credentials", category="auth")
railtownai.add_breadcrumb("Database query executed", category="database",
                         data={"query": "SELECT * FROM users", "duration_ms": 45})

# When an error occurs, all breadcrumbs are automatically attached
try:
    # Some operation that might fail
    result = risky_operation()
except Exception:
    logging.exception("Operation failed")  # This will include all the breadcrumbs above

Agent Observability

Track and monitor your AI agent executions with structured data upload. This feature allows you to store detailed information about agent runs, including nodes, steps, and execution flow.

1. Sign up for Conductr AI for FREE
2. Initialize Conductr AI with your API key in your Project Configuration (Logs)
3. Structure your agent data in the session format
4. Upload using upload_agent_run()

import json
import logging

import railtownai
import railtracks as rt
from fastapi import FastAPI, Query

logger = logging.getLogger(__name__)

# Initialize SDK
railtownai.init("YOUR_API_KEY")

app = FastAPI()

# Replace with your actual agent
# from your_agents_module import WeatherAgent


@app.get("/weather")
async def get_weather(
    city: str = Query(..., description="City name like Vancouver"),
    units: str = Query("metric", description="metric or imperial"),
):
    try:
        # Build the message history for the weather request
        message_history = rt.llm.MessageHistory([
            rt.llm.UserMessage(f"Weather request\nCity: {city}\nUnits: {units}")
        ])

        # Call the agent
        with rt.Session(name="agent-session") as session:
            result = await rt.call(WeatherAgent, message_history)

            # Upload agent run data to RailTown AI
            agent_run_data = session.payload()
            success = railtownai.upload_agent_run(agent_run_data)

            if success:
                logger.info("Agent run data uploaded successfully")
            else:
                logger.error("Failed to upload agent run data")

        logger.info("Weather processing completed successfully")

        return {
            "success": True,
            "city": city,
            "units": units,
            "analysis": str(result),
        }

    except Exception as e:
        logger.error(f"Error processing weather request: {e}")
        return {"error": str(e)}

If you are using the Railtown AI Python Logger, RailTracks Frameworkr automatically propagates any errors at run-time and attaches the node_id, run_id, and session_id via the python logging package, so that Conductr Agent Observability platform can show you exactly which nodes failed or retried.

Enterprise Connectors

Enterprise Connectors allows your agents to seamlessly integrate with services like Gmail, Slack, JIRA, Azure DevOps, and many more without having to manage the OAuth flow. Connect your integrations once on Conductr and your agents will be able to securely access your enterprise data.

Basic Example

import railtownai

async def main():
    data = await railtownai.connectors.get("gmail", "gmail/v1/users/me/messages", params={"maxResults": 3})
    messages = data["messages"]
    first_message = await railtownai.connectors.get("gmail", f"gmail/v1/users/me/messages/{messages[0]['id']}")

    print(first_message["payload"]["body"]["data"])

Agent Evaluations

Send agent evaluation results to Railengine (Conductr Agent Evaluations) for tracking and analysis.

1. Get your EVALUATIONS_API_TOKEN from the Conductr Agent Evaluations page
2. Add to .env: EVALUATIONS_API_TOKEN=<base64-encoded-json-from-conductr>
3. Call upload_agent_evaluation() with your evaluation payload

import railtownai

# Set EVALUATIONS_API_TOKEN in .env (from Conductr Agent Evaluations onboarding)
success = railtownai.upload_agent_evaluation(result_from_evaluation_run)

End-to-end: evaluate runs pulled from Conductr

Fetch the runs you have stored in Conductr with get_agent_runs() (see Fetching Agent Runs from Conductr) and pass the payloads straight into the railtracks evaluations API:

import railtownai
from railtracks import evaluations as evals

payloads = railtownai.get_agent_runs(agent_run_ids)
data = evals.extract_agent_data_points(payloads)   # railtracks >= the version with list[dict] support
results = evals.evaluate(data=data, evaluators=[...])
railtownai.upload_agent_evaluation(results)

Fetching Agent Runs from Conductr

Pull agent runs from the Conductr platform by id — for example when the platform triggers an evaluation against your deployed agent and hands it a list of agentRunIds.

1. Generate a project PAT in the platform UI (project → Secret Tokens)
2. Add to .env (or the environment):
   • CONDUCTR_PROJECT_PAT=<your-project-pat>
   • CONDUCTR_PROJECT_ID=<your-project-id>
   • RAILTOWN_API_URL=<platform-base-url> (optional; defaults to https://cndr.railtown.ai)
3. Call get_agent_runs()

railtownai.init() is not required when both project environment variables are set. The platform host defaults to https://cndr.railtown.ai; set RAILTOWN_API_URL to target a different environment, or call init() to have the host derived from your API key. CONDUCTR_PROJECT_ID overrides the project id from an init() key.

import railtownai

# Fetch one or more runs — pass a list of size 1 for a single run.
# Fail-fast by default; skip_errors=True logs a warning per failed run
# and returns the successful payloads instead.
payloads = railtownai.get_agent_runs([id1, id2], skip_errors=False)

Fetch failures raise railtownai.AgentRunFetchError (with .agent_run_id and .status_code); a missing PAT or project id raises railtownai.AgentRunsNotInitializedError.

Contributing

See the contributing guide for more information (including pip install -e ".[test,dev]" for tests and Ruff in a local clone).

License

The MIT License is a permissive license that allows you to:

• Use the software for any purpose
• Modify the software
• Distribute the software
• Use it commercially
• Use it privately
• Sublicense it

The only requirement is that the original copyright notice and license must be included in all copies or substantial portions of the software.