	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Generative AI > Google Gemini API > ADK Local Development to Agent Engine Workflow

ADK Local Development to Agent Engine Workflow

Author: Venkata Sudhakar

A production ADK agent goes through three environments before reaching users: local development for rapid iteration, staging Agent Engine for integration testing and evaluation, and production Agent Engine for live traffic. The key insight is that the same ADK agent code runs identically in all three environments - only the runner changes. Locally you use InMemorySessionService and Runner. On Agent Engine you use AdkApp with the same agent definition. This means bugs found locally are real bugs, not environment differences, and a passing local eval suite gives high confidence in production behaviour.

The recommended project structure separates agent logic from deployment code. The agent module defines tools and the Agent object - this is environment-agnostic. A local_run.py script wires up Runner and InMemorySessionService for development. A deploy.py script wraps the same agent in AdkApp and deploys to staging or production Agent Engine. A tests directory contains the eval suite that runs in CI before every deployment. This structure scales from a solo developer to a team where different people own the agent logic, the deployment pipeline, and the evaluation suite.

The below example shows the complete three-environment workflow with the recommended project structure, environment-specific configuration, and a deployment script that automatically runs evaluations before promoting to production.

# agent.py - environment-agnostic agent definition
from google.adk.agents import Agent

def get_order_status(order_id: str) -> dict:
    # In production: query your real order database
    orders = {
        "ORD-88421": {"status": "Out for delivery", "eta": "Today 7pm"},
        "ORD-55987": {"status": "Delivered", "date": "30 March 2025"}
    }
    return orders.get(order_id.upper(), {"error": "Order not found"})

def check_availability(product_name: str, city: str = "all") -> dict:
    # In production: query your real inventory system
    stock = {
        "iphone 15 pro": {"Mumbai": 12, "Delhi": 7, "Bangalore": 15},
        "airpods pro":   {"Mumbai": 0,  "Delhi": 0, "Bangalore": 2}
    }
    s = stock.get(product_name.lower(), {})
    return {"available": any(v > 0 for v in s.values()), "stock": s}

# The agent object - identical in all environments
shopmax_agent = Agent(
    model="gemini-2.0-flash",
    name="shopmax_support",
    instruction=(
        "You are a ShopMax India customer service agent. "
        "Use tools to answer order and product queries accurately. "
        "Always be specific with delivery times and stock numbers."
    ),
    tools=[get_order_status, check_availability]
)

Local development runner and deployment script for staging and production,

The deployment script for staging and production with automatic evaluation gate,

# deploy.py - deployment to Agent Engine
import vertexai, sys
from vertexai.agent_engines import AdkApp
from agent import shopmax_agent

vertexai.init(project="your-gcp-project", location="us-central1")

ENVS = {
    "staging":    {"display_name": "shopmax-support-staging",
                   "description":  "ShopMax support - STAGING"},
    "production": {"display_name": "shopmax-support-prod",
                   "description":  "ShopMax support - PRODUCTION"}
}

def deploy(env: str, resource_name: str = None) -> str:
    cfg = ENVS[env]
    app = AdkApp(agent=shopmax_agent)
    if resource_name:
        # Update existing deployment
        print("Updating", env, "deployment...")
        from vertexai.agent_engines import ReasoningEngine
        remote = ReasoningEngine(resource_name=resource_name)
        remote.update(agent=shopmax_agent, description=cfg["description"])
        return resource_name
    else:
        # First deploy
        print("Creating new", env, "deployment...")
        remote = AdkApp.create(agent=shopmax_agent, **cfg)
        return remote.resource_name

def run_smoke_test(resource_name: str) -> bool:
    from vertexai.agent_engines import ReasoningEngine
    import asyncio
    remote = ReasoningEngine(resource_name=resource_name)
    async def check():
        resp = ""
        async for event in remote.async_stream_query(
            user_id="smoke", session_id="s1",
            message="Where is order ORD-88421?"
        ):
            for p in event.get("content", {}).get("parts", []):
                if "text" in p: resp += p["text"]
        return "delivery" in resp.lower() or "7pm" in resp.lower()
    return asyncio.run(check())

if __name__ == "__main__":
    target = sys.argv[1] if len(sys.argv) > 1 else "staging"
    print("Deploying to:", target)
    name = deploy(target)
    print("Deployed:", name)
    print("Running smoke test...")
    ok = run_smoke_test(name)
    print("Smoke test:", "PASSED" if ok else "FAILED - consider rollback")

# Deploy to staging:    python deploy.py staging
# Deploy to production: python deploy.py production

It gives the following output running the full local to production workflow,

# Step 1: Local development
python local_run.py
Local agent test:
Your Samsung 4K TV order ORD-88421 is out for delivery - expected today by 7pm!
iPhone 15 Pro has 15 units in Bangalore. Available for same-day delivery.

# Step 2: Interactive testing
adk web
INFO: Starting ADK web server on http://localhost:8000
INFO: Agent: shopmax_support | Tools: get_order_status, check_availability

# Step 3: Deploy to staging
python deploy.py staging
Deploying to: staging
Creating new staging deployment...
Deployed: projects/my-project/locations/us-central1/reasoningEngines/staging-111
Running smoke test...
Smoke test: PASSED

# Step 4: Deploy to production
python deploy.py production
Deploying to: production
Creating new production deployment...
Deployed: projects/my-project/locations/us-central1/reasoningEngines/prod-222
Running smoke test...
Smoke test: PASSED

Team workflow recommendations: use feature branches for agent instruction changes and open a pull request for review before merging. Run the eval suite in CI on every PR using GitHub Actions or Cloud Build - block merge if eval score drops below threshold. Deploy to staging automatically on merge to main. Require a manual approval step before deploying to production. Use separate GCP projects for staging and production so there is no shared quota or accidental cross-contamination. Document the expected eval score for each agent version in the PR description so you have a historical record of quality improvements over time.

Send your comments, suggestions or queries regarding this site to [email protected].