Skip to content

Agents API Reference

cs_copilot.agents

Factories and team builders for ChemSpace Copilot agents.

The registry contains nine agent types. Seven are members of the production team: ChEMBL retrieval, GTM, chemoinformatics, reporting, molecular design, peptide design, and retrosynthesis. robustness_evaluation is a separate evaluation agent, while single_agent is the flat baseline used by architecture-ablation tests.

Use :func:create_agent to construct one registered type, :func:get_cs_copilot_agent_team for the production team, or :func:get_cs_copilot_single_agent for the controlled baseline.

Exports are resolved lazily so that importing a lightweight submodule such as :mod:cs_copilot.agents.instructions (used verbatim by the MCP prompt registry) does not construct the Agno team, factory, or registry modules.

AgentConfig dataclass

Configuration for creating an agent.

Source code in src/cs_copilot/agents/factories.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@dataclass
class AgentConfig:
    """Configuration for creating an agent."""

    name: str
    description: str
    tools: List[Any] = field(default_factory=list)
    instructions: List[str] = field(default_factory=list)
    session_state: Dict[str, Any] = field(default_factory=dict)
    role_policy: Optional[RolePolicy] = None

    def validate(self) -> None:
        """Validate the agent configuration."""
        if not self.name:
            raise ValueError("Agent name cannot be empty")
        if not self.description:
            raise ValueError("Agent description cannot be empty")
        if not isinstance(self.tools, list):
            raise TypeError("Tools must be a list")
        if not isinstance(self.instructions, list):
            raise TypeError("Instructions must be a list")
        if self.role_policy is not None:
            validate_role_tools(self.role_policy, self.tools)

validate()

Validate the agent configuration.

Source code in src/cs_copilot/agents/factories.py 
82
83
84
85
86
87
88
89
90
91
92
93
def validate(self) -> None:
    """Validate the agent configuration."""
    if not self.name:
        raise ValueError("Agent name cannot be empty")
    if not self.description:
        raise ValueError("Agent description cannot be empty")
    if not isinstance(self.tools, list):
        raise TypeError("Tools must be a list")
    if not isinstance(self.instructions, list):
        raise TypeError("Instructions must be a list")
    if self.role_policy is not None:
        validate_role_tools(self.role_policy, self.tools)

AgentCreationError

Bases: Exception

Exception raised when agent creation fails.

Source code in src/cs_copilot/agents/factories.py 
96
97
98
99
class AgentCreationError(Exception):
    """Exception raised when agent creation fails."""

    pass

BaseAgentFactory

Bases: ABC

Base class for creating agents with common configuration and error handling.

Source code in src/cs_copilot/agents/factories.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
class BaseAgentFactory(ABC):
    """Base class for creating agents with common configuration and error handling."""

    def __init__(self, logger: Optional[logging.Logger] = None):
        self.logger = logger or logging.getLogger(__name__)

    @abstractmethod
    def get_agent_config(self) -> AgentConfig:
        """Return the configuration for this agent type."""
        pass

    def create_agent(
        self,
        model: Model,
        markdown: bool = True,
        debug_mode: bool = False,
        enable_mlflow_tracking: bool = True,
        **kwargs: Any,
    ) -> Agent:
        """Create an agent with error handling and validation.

        Args:
            model: Model to use for the agent
            markdown: Whether to enable markdown formatting
            debug_mode: Whether to enable debug mode
            enable_mlflow_tracking: Whether to enable MLflow tracking for this agent
            **kwargs: Additional keyword arguments for agent creation

        Returns:
            Created agent instance
        """
        try:
            config = self.get_agent_config()
            agent_type = getattr(self.__class__, "agent_type", None)
            declared_policy = ROLE_POLICIES.get(agent_type)
            if config.role_policy is None:
                config.role_policy = declared_policy
            elif declared_policy is not None and config.role_policy != declared_policy:
                raise ValueError(
                    f"Factory role policy for '{agent_type}' differs from the canonical allowlist"
                )
            config.validate()
            provided_session_state = kwargs.pop("session_state", None)

            # Log agent creation
            self.logger.info(f"Creating agent: {config.name}")

            # Create agent with common parameters
            agent_kwargs = {
                "model": model,
                "name": config.name,
                "description": config.description,
                "tools": config.tools,
                "markdown": markdown,
                "debug_mode": debug_mode,
                "enable_agentic_state": True,
                "add_session_state_to_context": True,
            }

            # Add optional parameters if they exist
            if config.instructions:
                agent_kwargs["instructions"] = config.instructions
            if provided_session_state is not None:
                if config.session_state:
                    _merge_session_state_defaults(provided_session_state, config.session_state)
                agent_kwargs["session_state"] = provided_session_state
            elif config.session_state:
                agent_kwargs["session_state"] = config.session_state

            # Add any additional kwargs passed in
            agent_kwargs.update(kwargs)

            agent = Agent(**agent_kwargs)
            if config.role_policy is not None:
                # Keep the enforced policy inspectable without placing it in the
                # shared session state (where one member could overwrite another).
                agent.role_policy = config.role_policy

            # Wrap agent methods with MLflow tracking if enabled
            if enable_mlflow_tracking:
                agent = self._wrap_agent_with_tracking(agent, config)

            self.logger.info(f"Successfully created agent: {config.name}")
            return agent

        except Exception as e:
            self.logger.error(
                f"Failed to create agent {config.name if 'config' in locals() else 'unknown'}: {str(e)}"
            )
            raise AgentCreationError(f"Failed to create agent: {str(e)}") from e

    def _wrap_agent_with_tracking(self, agent: Agent, config: AgentConfig) -> Agent:
        """Wrap agent execution methods with MLflow tracking.

        Args:
            agent: Agent instance to wrap
            config: Agent configuration

        Returns:
            Agent with wrapped methods
        """
        try:
            from cs_copilot.tracking import get_tracker
            from cs_copilot.tracking.utils import build_prompt_signature

            tracker = get_tracker()

            if not tracker.is_enabled():
                return agent

            # Get the agent type from the factory
            agent_type = getattr(self.__class__, "agent_type", None)

            def build_prompt_template() -> Optional[str]:
                sections = []
                if config.description:
                    sections.append(str(config.description).strip())
                if config.instructions:
                    normalized = [
                        str(item).strip() for item in config.instructions if item is not None
                    ]
                    instructions_text = "\n".join(normalized).strip()
                    if instructions_text:
                        sections.append(instructions_text)
                template = "\n\n".join([section for section in sections if section])
                return template.strip() if template else None

            def build_prompt_name() -> str:
                base_name = agent_type or config.name
                safe_name = str(base_name).replace(" ", "_").lower()
                return f"cs_copilot.{safe_name}"

            prompt_template = build_prompt_template()
            prompt_signature = build_prompt_signature(prompt_template)
            prompt_registry_name = build_prompt_name()

            def register_prompt_in_registry():
                if not prompt_template:
                    return
                commit_message = None
                if prompt_signature:
                    commit_message = f"cs_copilot auto update ({prompt_signature.version})"
                prompt_obj = tracker.register_prompt_version(
                    name=prompt_registry_name,
                    template=prompt_template,
                    commit_message=commit_message,
                    tags={
                        "agent_name": agent.name,
                        "agent_type": agent_type or "unknown",
                        "component": "cs_copilot",
                    },
                )
                if prompt_obj:
                    version = getattr(prompt_obj, "version", None)
                    tracker.log_params(
                        {
                            "prompt_registry_name": prompt_registry_name,
                            "prompt_registry_version": str(version) if version is not None else "",
                            "prompt_registry_uri": (
                                f"prompts:/{prompt_registry_name}/{version}"
                                if version is not None
                                else ""
                            ),
                        }
                    )

            # Wrap run() method
            original_run = agent.run

            def tracked_run(*args, **kwargs):
                # Extract prompt from args
                prompt = args[0] if args else kwargs.get("message", "")

                with tracker.track_agent_run(
                    agent_name=agent.name, prompt=str(prompt), agent_type=agent_type
                ):
                    # Log agent configuration
                    tracker.log_params(
                        {
                            "agent_name": agent.name,
                            "agent_type": agent_type or "unknown",
                            "num_tools": len(config.tools),
                            "tools": ",".join([t.__class__.__name__ for t in config.tools]),
                        }
                    )
                    register_prompt_in_registry()

                    result = original_run(*args, **kwargs)

                    # Log result metrics if available
                    if hasattr(result, "content") and result.content:
                        from cs_copilot.tracking.utils import count_tokens

                        tracker.log_metrics(
                            {"output_tokens_estimate": float(count_tokens(result.content))}
                        )

                    return result

            agent.run = tracked_run

            # Wrap arun() method (async version)
            original_arun = agent.arun

            async def tracked_arun(*args, **kwargs):
                # Extract prompt from args
                prompt = args[0] if args else kwargs.get("message", "")

                with tracker.track_agent_run(
                    agent_name=agent.name, prompt=str(prompt), agent_type=agent_type
                ):
                    # Log agent configuration
                    tracker.log_params(
                        {
                            "agent_name": agent.name,
                            "agent_type": agent_type or "unknown",
                            "num_tools": len(config.tools),
                            "tools": ",".join([t.__class__.__name__ for t in config.tools]),
                        }
                    )
                    register_prompt_in_registry()

                    result = await original_arun(*args, **kwargs)

                    # Log result metrics if available
                    if hasattr(result, "content") and result.content:
                        from cs_copilot.tracking.utils import count_tokens

                        tracker.log_metrics(
                            {"output_tokens_estimate": float(count_tokens(result.content))}
                        )

                    return result

            agent.arun = tracked_arun

            self.logger.debug(f"MLflow tracking enabled for agent: {agent.name}")

        except ImportError:
            self.logger.warning(
                "MLflow tracking module not available. Agent will run without tracking."
            )
        except Exception as e:
            self.logger.warning(f"Failed to enable MLflow tracking for agent: {e}")

        return agent

get_agent_config() abstractmethod

Return the configuration for this agent type.

Source code in src/cs_copilot/agents/factories.py 
118
119
120
121
@abstractmethod
def get_agent_config(self) -> AgentConfig:
    """Return the configuration for this agent type."""
    pass

create_agent(model, markdown=True, debug_mode=False, enable_mlflow_tracking=True, **kwargs)

Create an agent with error handling and validation.

Parameters:

Name Type Description Default
model Model

Model to use for the agent

 required
markdown bool

Whether to enable markdown formatting

 True
debug_mode bool

Whether to enable debug mode

 False
enable_mlflow_tracking bool

Whether to enable MLflow tracking for this agent

 True
**kwargs Any

Additional keyword arguments for agent creation

 {}

Returns:

Type Description
Agent

Created agent instance
Source code in src/cs_copilot/agents/factories.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def create_agent(
    self,
    model: Model,
    markdown: bool = True,
    debug_mode: bool = False,
    enable_mlflow_tracking: bool = True,
    **kwargs: Any,
) -> Agent:
    """Create an agent with error handling and validation.

    Args:
        model: Model to use for the agent
        markdown: Whether to enable markdown formatting
        debug_mode: Whether to enable debug mode
        enable_mlflow_tracking: Whether to enable MLflow tracking for this agent
        **kwargs: Additional keyword arguments for agent creation

    Returns:
        Created agent instance
    """
    try:
        config = self.get_agent_config()
        agent_type = getattr(self.__class__, "agent_type", None)
        declared_policy = ROLE_POLICIES.get(agent_type)
        if config.role_policy is None:
            config.role_policy = declared_policy
        elif declared_policy is not None and config.role_policy != declared_policy:
            raise ValueError(
                f"Factory role policy for '{agent_type}' differs from the canonical allowlist"
            )
        config.validate()
        provided_session_state = kwargs.pop("session_state", None)

        # Log agent creation
        self.logger.info(f"Creating agent: {config.name}")

        # Create agent with common parameters
        agent_kwargs = {
            "model": model,
            "name": config.name,
            "description": config.description,
            "tools": config.tools,
            "markdown": markdown,
            "debug_mode": debug_mode,
            "enable_agentic_state": True,
            "add_session_state_to_context": True,
        }

        # Add optional parameters if they exist
        if config.instructions:
            agent_kwargs["instructions"] = config.instructions
        if provided_session_state is not None:
            if config.session_state:
                _merge_session_state_defaults(provided_session_state, config.session_state)
            agent_kwargs["session_state"] = provided_session_state
        elif config.session_state:
            agent_kwargs["session_state"] = config.session_state

        # Add any additional kwargs passed in
        agent_kwargs.update(kwargs)

        agent = Agent(**agent_kwargs)
        if config.role_policy is not None:
            # Keep the enforced policy inspectable without placing it in the
            # shared session state (where one member could overwrite another).
            agent.role_policy = config.role_policy

        # Wrap agent methods with MLflow tracking if enabled
        if enable_mlflow_tracking:
            agent = self._wrap_agent_with_tracking(agent, config)

        self.logger.info(f"Successfully created agent: {config.name}")
        return agent

    except Exception as e:
        self.logger.error(
            f"Failed to create agent {config.name if 'config' in locals() else 'unknown'}: {str(e)}"
        )
        raise AgentCreationError(f"Failed to create agent: {str(e)}") from e

create_agent(agent_type, model, **kwargs)

Create an agent by type using the global registry.

Parameters:

Name Type Description Default
agent_type str

The type of agent to create

 required
model Model

The language model to use

 required
**kwargs Any

Additional arguments passed to the agent factory

 {}

Returns:

Name Type Description
Agent Agent

The created agent instance

Raises:

Type Description
ValueError

If agent_type is not registered
AgentCreationError

If agent creation fails
Source code in src/cs_copilot/agents/registry.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def create_agent(agent_type: str, model: Model, **kwargs: Any) -> Agent:
    """
    Create an agent by type using the global registry.

    Args:
        agent_type: The type of agent to create
        model: The language model to use
        **kwargs: Additional arguments passed to the agent factory

    Returns:
        Agent: The created agent instance

    Raises:
        ValueError: If agent_type is not registered
        AgentCreationError: If agent creation fails
    """
    return _agent_registry.create_agent(agent_type, model, **kwargs)

get_registry()

Get the global agent registry instance.

Source code in src/cs_copilot/agents/registry.py 
123
124
125
def get_registry() -> AgentRegistry:
    """Get the global agent registry instance."""
    return _agent_registry

list_available_agent_types()

List all available agent types.

Source code in src/cs_copilot/agents/registry.py 
118
119
120
def list_available_agent_types() -> List[str]:
    """List all available agent types."""
    return _agent_registry.list_agent_types()

get_cs_copilot_single_agent(model, *, markdown=True, debug_mode=False, enable_mlflow_tracking=True)

Create the single flat agent (all toolkits, no team, no routing).

Parameters:

Name Type Description Default
model Model

Agno Model instance — pass the SAME instance used for the team so the ablation holds the model constant.

 required
markdown bool

Format output in markdown (matches team members).

 True
debug_mode bool

Enable debug logs.

 False
enable_mlflow_tracking bool

Wrap run/arun with MLflow tracking, exactly as team members are wrapped, for a fair comparison.

 True

Returns:

Name Type Description
Agent Agent

The configured single-agent baseline. Exposes .run /
Agent

.get_session_state like the team, so the robustness harness can drive
Agent

either arm through one interface.
Source code in src/cs_copilot/agents/single_agent.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def get_cs_copilot_single_agent(
    model: Model,
    *,
    markdown: bool = True,
    debug_mode: bool = False,
    enable_mlflow_tracking: bool = True,
) -> Agent:
    """Create the single flat agent (all toolkits, no team, no routing).

    Args:
        model: Agno Model instance — pass the SAME instance used for the team so
            the ablation holds the model constant.
        markdown: Format output in markdown (matches team members).
        debug_mode: Enable debug logs.
        enable_mlflow_tracking: Wrap ``run``/``arun`` with MLflow tracking, exactly
            as team members are wrapped, for a fair comparison.

    Returns:
        Agent: The configured single-agent baseline. Exposes ``.run`` /
        ``.get_session_state`` like the team, so the robustness harness can drive
        either arm through one interface.
    """
    logger = logging.getLogger(__name__)
    logger.info("Creating Cs_copilot single-agent baseline")

    # Seed the same shared-state contract the team members get (teams.py:70-84):
    # resource profile + a scratch dict. The factory's union session_state is then
    # deep-merged in by BaseAgentFactory.create_agent without overwriting these.
    seed_session_state = {
        "resource_profile": analyze_resources(),
        "agent_scratch": {},
    }

    agent = create_agent(
        "single_agent",
        model=model,
        markdown=markdown,
        debug_mode=debug_mode,
        enable_mlflow_tracking=enable_mlflow_tracking,
        session_state=seed_session_state,
    )
    logger.info("Successfully created Cs_copilot single-agent baseline")
    return agent

get_cs_copilot_agent_team(model, *, markdown=True, debug_mode=False, show_members_responses=True, enable_memory=True, db_file=None, enable_mlflow_tracking=True, run_context=None, execution_mode=None)

Create a coordinated team of cs_copilot agents using Agno.

Parameters:

Name Type Description Default
model Model

Agno Model instance used for team coordination and member agents

 required
markdown bool

Format output in markdown

 True
debug_mode bool

Enable debug logs

 False
show_members_responses bool

Print member responses during coordination

 True
enable_memory bool

Enable persistent session history (default: True). Cross-session user/agentic memories stay disabled to prevent state leakage.

 True
db_file str

Custom database file path. If not provided, uses CS_COPILOT_MEMORY_DB. Use unique paths for session isolation in testing.

 None
enable_mlflow_tracking bool

Enable MLflow tracking for agents (default: True). Set to False to disable tracking.

 True
run_context Any

Optional v2 workflow RunContext used to record validated specialist handoffs. When it is a real RunContext, the team's tool calls are routed through the execution kernel immediately; otherwise ensure_agno_session_run attaches the chat's run before the first message.

 None
execution_mode ExecutionMode | str | None

off, observe, or enforce. Defaults to CS_COPILOT_AGNO_EXECUTION (enforce when unset).

 None

Returns:

Name Type Description
Team Team

Configured Cs_copilot team

Raises:

Type Description
AgentCreationError

If one or more agents fail to initialize
Source code in src/cs_copilot/agents/teams.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def get_cs_copilot_agent_team(
    model: Model,  # Agno Model instance, e.g. OpenAIChat(...) or Claude(...)
    *,
    markdown: bool = True,
    debug_mode: bool = False,
    show_members_responses: bool = True,
    enable_memory: bool = True,
    db_file: str = None,
    enable_mlflow_tracking: bool = True,
    run_context: Any = None,
    execution_mode: ExecutionMode | str | None = None,
) -> Team:
    """
    Create a coordinated team of cs_copilot agents using Agno.

    Args:
        model: Agno Model instance used for team coordination and member agents
        markdown: Format output in markdown
        debug_mode: Enable debug logs
        show_members_responses: Print member responses during coordination
        enable_memory: Enable persistent session history (default: True). Cross-session
                      user/agentic memories stay disabled to prevent state leakage.
        db_file: Custom database file path. If not provided, uses CS_COPILOT_MEMORY_DB.
                Use unique paths for session isolation in testing.
        enable_mlflow_tracking: Enable MLflow tracking for agents (default: True).
                               Set to False to disable tracking.
        run_context: Optional v2 workflow ``RunContext`` used to record validated
                     specialist handoffs. When it is a real ``RunContext``, the
                     team's tool calls are routed through the execution kernel
                     immediately; otherwise ``ensure_agno_session_run`` attaches
                     the chat's run before the first message.
        execution_mode: ``off``, ``observe``, or ``enforce``. Defaults to
                     ``CS_COPILOT_AGNO_EXECUTION`` (``enforce`` when unset).

    Returns:
        Team: Configured Cs_copilot team

    Raises:
        AgentCreationError: If one or more agents fail to initialize
    """
    logger = logging.getLogger(__name__)
    logger.info("Creating Cs_copilot Agent Team")

    # ✅ Single DB handles session storage/history in v2.1.x.
    # Cross-session memories are intentionally disabled below; only per-thread
    # history/session state should persist.
    db = None
    if enable_memory:
        db = SqliteDb(
            db_file=db_file
            or CS_COPILOT_MEMORY_DB
            # NOTE: CS_COPILOT_MEMORY_TABLE is not required by SqliteDb.
            # Agno manages its own tables for sessions/memories. Kept import for compat.
        )

    # Probe runtime environment (GPU, CPU, RAM, databases, cached models)
    resource_profile = analyze_resources()
    logger.info("Resource profile: %s", resource_profile)
    coordinator_tools = [SessionMemoryToolkit(), SkillToolkit()]
    validate_role_tools(get_role_policy(COORDINATOR_ROLE), coordinator_tools)
    agentic_contracts = {
        "handoff_schema_version": HANDOFF_SCHEMA_VERSION,
        "context_budget": DEFAULT_CONTEXT_BUDGET.as_dict(),
        "delegation_limits": DEFAULT_DELEGATION_LIMITS.as_dict(),
        "handoff_transport": "delegate_task_to_member.task_description JSON",
        "role_profiles": {
            role: policy.profile for role, policy in ROLE_POLICIES.items() if role != "single_agent"
        },
    }
    active_workflow_run = getattr(run_context, "run", None)
    if active_workflow_run is not None:
        active_run = {
            "run_id": active_workflow_run.run_id,
            "workflow_slug": active_workflow_run.workflow_slug,
            "trace_id": active_workflow_run.trace_id,
        }
    else:
        # Even ad-hoc Agno sessions need canonical handoff identity. This is
        # deliberately process-local unless a durable RunContext is supplied.
        active_run = {
            "run_id": f"agno-{uuid.uuid4().hex[:12]}",
            "workflow_slug": "ad-hoc",
            "trace_id": uuid.uuid4().hex,
        }
    agentic_contracts["active_run"] = active_run
    shared_session_state = {
        "resource_profile": resource_profile,
        "agent_scratch": {},
        "agentic_contracts": agentic_contracts,
        "current_run_id": active_run["run_id"],
    }

    # Common agent parameters supplied by the factory
    agent_params = {
        "markdown": markdown,
        "debug_mode": debug_mode,
        "enable_mlflow_tracking": enable_mlflow_tracking,
        "session_state": shared_session_state,
    }

    # ============================================================================
    # 7-AGENT PRODUCTION TEAM
    # ============================================================================
    # Consolidation history:
    #   MERGED: GTM Optimization + Loading + Density + Activity → GTM Agent
    #   GENERALIZED: GTM Chemotype Analysis → Chemoinformatician (method-agnostic)
    #   TRANSFORMED: Autoencoder public agent → Molecular Designer (engine-based)
    #   ADDED: Report Generator (presentation layer)
    #   REMOVED: Robustness Evaluator (not included in main team, invoked separately)
    # ============================================================================

    # (type_key, human_name)
    agents_config: List[Tuple[str, str]] = [
        ("chembl_downloader", "ChEMBL Downloader"),
        (
            "gtm_agent",
            "GTM Agent",
        ),  # Unified GTM operations (build, load, density, activity, project)
        (
            "chemoinformatician",
            "Chemoinformatician",
        ),  # Comprehensive chemoinformatics (chemotype, clustering, SAR, similarity, QSAR)
        ("report_generator", "Report Generator"),  # Universal presentation layer
        ("molecular_designer", "Molecular Designer"),  # Small-molecule design engines
        ("peptide_designer", "Peptide Designer"),  # Peptide design engines
        ("synplanner", "SynPlanner"),
        # Note: Robustness Evaluator excluded from main team (invoked separately for testing)
    ]

    agents = []
    failures = []

    for agent_type, agent_name in agents_config:
        try:
            logger.info("Creating %s agent", agent_name)
            agent = create_agent(agent_type, model=model, **agent_params)
            agent.agentic_role = agent_type
            agents.append(agent)
            logger.info("Successfully created %s agent", agent_name)
        except Exception as e:
            logger.exception("Failed to create %s agent", agent_name)
            failures.append(f"{agent_name}: {e!s}")

    if failures:
        msg = "Agent initialization failures:\n  - " + "\n  - ".join(failures)
        raise AgentCreationError(msg)

    agentic_contracts["member_roles"] = {
        get_member_id(agent): agent.agentic_role for agent in agents
    }
    delegation_guard = StructuredDelegationGuard()
    team = StructuredHandoffTeam(
        name="Cs_copilot Team",
        members=agents,
        model=model,
        run_context=run_context,
        delegation_guard=delegation_guard,
        # ✅ Attach DB directly to the team (persists sessions/history)
        # If enable_memory=False, db=None prevents any persistence
        db=db,
        # Keep session history, but never inject cross-session memories. Agno
        # defaults add_memories_to_context=True when agentic memory is enabled,
        # which caused new chats to recall prior chemical-space analyses.
        enable_agentic_memory=False,
        enable_user_memories=False,
        add_memories_to_context=False,
        add_history_to_context=enable_memory,  # include recent history in prompts
        num_history_runs=5 if enable_memory else 0,  # 🔧 LIMIT context to last 5 runs
        # The coordinator passes bounded handoff envelopes; member transcripts
        # are not broadcast to peers as an implicit second context channel.
        share_member_interactions=False,
        store_history_messages=enable_memory,  # persist message history to DB
        store_tool_messages=enable_memory,  # persist tool results
        store_media=enable_memory,  # persist any media if used
        # Session state (always enabled for within-session data passing)
        session_state=shared_session_state,
        add_session_state_to_context=True,
        enable_agentic_state=True,
        tools=coordinator_tools,
        # Prompting
        description=(
            "You are an intelligent coordinator orchestrating a team of specialized cheminformatics agents. "
            "Your role is to understand user requests, select the appropriate agent(s) or workflows, "
            "and chain multiple agents when needed to complete complex analyses.\n\n"
            "• ChEMBL Downloader: Download bioactivity data from ChEMBL database\n"
            "• GTM Agent: All GTM operations (build/load/density/activity/project) with smart caching\n"
            "• Chemoinformatician: Downstream analysis (scaffold, SAR, similarity, clustering) - works with GTM output\n"
            "• Report Generator: Universal presentation layer for all analysis types\n"
            "• Molecular Designer: Small-molecule design via autoencoder and LLM engines (SMILES, standalone + GTM-guided)\n"
            "• Peptide Designer: Peptide design via WAE and LLM engines + GTM on latent space + DBAASP antimicrobial activity landscapes\n"
            "• SynPlanner: Retrosynthetic planning for target molecules\n\n"
            # Routing prose is generated from the workflow/skill catalog keywords
            # so it can never drift from the deterministic MCP bootstrap routing.
            + render_routing_rules() + "\n\n"
            "When coordinating: (1) Assess if a predefined workflow covers the request, (2) Select and chain "
            "specialized agents for multi-step tasks (GTM → Chemoinformatician → Report Generator is common), "
            "(3) For analysis requests, automatically add Report Generator unless user explicitly requests raw data only, "
            "(4) For ambiguous opening requests, apply the INITIAL CLARIFICATION FLOW (peptides vs molecules, then exploratory vs generative), (5) Synthesize insights from agent outputs into coherent analyses."
        ),
        instructions=[
            *AGENT_TEAM_INSTRUCTIONS,
            (
                "For multi-step scientific workflows, consult the Skills tools "
                "(`list_skills`, `search_skills`, `fetch_skill`) before routing "
                "specialized agents so the team follows reusable ChemSpace procedures."
            ),
            (
                "When calling `delegate_task_to_member`, `task_description` MUST be exactly "
                "one JSON object with schema_version, run_id, workflow_slug, task_id, "
                "sender_role, receiver_role, objective, constraints, required_capabilities, "
                "input_artifact_ids, expected_output_artifacts, expected_output_schema, "
                "acceptance_criteria, context_summary, budget (max_tokens, max_tool_calls, "
                "timeout_seconds), trace_id, span_id, and optional parent_span_id. Use the "
                "run_id, workflow_slug, and trace_id from agentic_contracts.active_run. Use the "
                "selected member's canonical role from agentic_contracts.member_roles as "
                "receiver_role. Never include private reasoning, scratchpads, messages, or "
                "conversation history. The delegation guard derives expected_output from "
                "this object and rejects unstructured, oversized, repeated, or looping calls."
            ),
        ],
        # UX & observability
        markdown=markdown,
        debug_mode=debug_mode,
        stream_member_events=True,  # stream events from members (Team API)
        show_members_responses=show_members_responses,
    )
    team.execution_mode = execution_mode_from_env(execution_mode)
    if (
        team.execution_mode is not ExecutionMode.OFF
        and hasattr(run_context, "bind_session_state")
        and hasattr(run_context, "layout")
    ):
        run_context.bind_session_state(shared_session_state)
        attach_execution(team, run_context=run_context, mode=team.execution_mode)
    logger.info("Successfully created Cs_copilot Agent Team")
    return team

get_last_agent_reply(agent)

Extract the content of the last message from an agent's session.

Source code in src/cs_copilot/agents/utils.py 
12
13
14
def get_last_agent_reply(agent: Agent) -> str:
    """Extract the content of the last message from an agent's session."""
    return copy.deepcopy(agent.get_messages_for_session()[-1].to_dict()["content"])

agno_compat

Compatibility fixes for the pinned Agno release.

Agno 2.1.9's Model.arun_function_call leaves its result unassigned when a tool's pre-hook raises an AgentRunException (RetryAgentRun / StopAgentRun), so the async team path crashes with UnboundLocalError instead of turning the exception into a retry request for the model, as the synchronous run_function_call does. The structured-handoff guard rejects malformed delegations exactly that way, so without this fix any rejected delegation in an async chat (Chainlit, the CLI) aborts the whole turn.

:func:patch_async_function_call_retry installs the upstream method with the single missing initialization. It only applies to the pinned version and is idempotent.

patch_async_function_call_retry()

Make async tool pre-hook retries behave like the synchronous path.

Returns whether the patch is active.

Source code in src/cs_copilot/agents/agno_compat.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def patch_async_function_call_retry() -> bool:
    """Make async tool pre-hook retries behave like the synchronous path.

    Returns whether the patch is active.
    """

    from agno.models.base import Model

    if getattr(Model.arun_function_call, _PATCH_MARKER, False):
        return True
    try:
        installed = version("agno")
    except PackageNotFoundError:  # pragma: no cover - agno is a hard dependency
        return False
    if installed not in PATCHED_AGNO_VERSIONS:
        logger.warning(
            "Not applying the async tool-retry fix for agno %s (validated for %s); "
            "check whether Model.arun_function_call still needs it.",
            installed,
            ", ".join(sorted(PATCHED_AGNO_VERSIONS)),
        )
        return False

    from inspect import isasyncgenfunction, iscoroutine, iscoroutinefunction

    from agno.exceptions import AgentRunException
    from agno.tools.function import FunctionExecutionResult
    from agno.utils.log import log_error
    from agno.utils.timer import Timer

    async def arun_function_call(self, function_call):
        """Run a single function call and return its success status, timer, and the FunctionCall object."""

        function_call_timer = Timer()
        function_call_timer.start()
        success = False
        # The only change from agno 2.1.9: a pre-hook AgentRunException must
        # still return a result, exactly like the synchronous path.
        result = FunctionExecutionResult(status="failure")

        try:
            if (
                iscoroutinefunction(function_call.function.entrypoint)
                or isasyncgenfunction(function_call.function.entrypoint)
                or iscoroutine(function_call.function.entrypoint)
            ):
                result = await function_call.aexecute()
                success = result.status == "success"

            # If any of the hooks are async, we need to run the function call asynchronously
            elif function_call.function.tool_hooks is not None and any(
                iscoroutinefunction(f) for f in function_call.function.tool_hooks
            ):
                result = await function_call.aexecute()
                success = result.status == "success"
            else:
                import asyncio

                result = await asyncio.to_thread(function_call.execute)
                success = result.status == "success"
        except AgentRunException as e:
            success = e
        except Exception as e:
            log_error(f"Error executing function {function_call.function.name}: {e}")
            success = False
            raise e

        function_call_timer.stop()
        return success, function_call_timer, function_call, result

    setattr(arun_function_call, _PATCH_MARKER, True)
    Model.arun_function_call = arun_function_call
    return True

patch_tool_result_coercion()

Stop Agno truth-testing a tool result it cannot truth-test.

Agno 2.1.9 renders every tool result with str(result) if result else "" (agno/models/base.py). A pandas DataFrame raises ValueError there, aborting a run the agent had already largely completed, and a NumPy array raises the same way.

Patching :meth:FunctionCall.execute covers every caller -- the team path, the flat single-agent baseline, and tool calls a member makes while a delegation is in flight -- because that is where Agno collects the value. The execution kernel has already seen the raw value by then, so artifact registration is unaffected.

Returns whether the patch is active.

Source code in src/cs_copilot/agents/agno_compat.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def patch_tool_result_coercion() -> bool:
    """Stop Agno truth-testing a tool result it cannot truth-test.

    Agno 2.1.9 renders every tool result with
    ``str(result) if result else ""`` (``agno/models/base.py``). A pandas
    DataFrame raises ``ValueError`` there, aborting a run the agent had already
    largely completed, and a NumPy array raises the same way.

    Patching :meth:`FunctionCall.execute` covers every caller -- the team path,
    the flat single-agent baseline, and tool calls a member makes while a
    delegation is in flight -- because that is where Agno collects the value.
    The execution kernel has already seen the raw value by then, so artifact
    registration is unaffected.

    Returns whether the patch is active.
    """

    from agno.tools.function import FunctionCall

    if getattr(FunctionCall.execute, _RESULT_PATCH_MARKER, False):
        return True
    try:
        installed = version("agno")
    except PackageNotFoundError:  # pragma: no cover - agno is a hard dependency
        return False
    if installed not in PATCHED_AGNO_VERSIONS:
        logger.warning(
            "Not applying the tool-result coercion fix for agno %s (validated for %s); "
            "check whether run_function_call still truth-tests its result.",
            installed,
            ", ".join(sorted(PATCHED_AGNO_VERSIONS)),
        )
        return False

    from cs_copilot.execution.envelopes import _coerce_return_value

    def _truth_testable(value) -> bool:
        try:
            bool(value)
        except Exception:
            return False
        return True

    def _coerce(outcome):
        value = getattr(outcome, "result", None)
        if value is None or _truth_testable(value):
            return outcome
        try:
            coerced = _coerce_return_value(value)
        except Exception:  # never fail a completed call over rendering
            coerced = value
        if not _truth_testable(coerced):
            # Anything else whose __bool__ raises, such as a NumPy array.
            coerced = coerced.tolist() if hasattr(coerced, "tolist") else repr(coerced)
        outcome.result = coerced
        return outcome

    original_execute = FunctionCall.execute
    original_aexecute = FunctionCall.aexecute

    def execute(self, *args, **kwargs):
        return _coerce(original_execute(self, *args, **kwargs))

    async def aexecute(self, *args, **kwargs):
        return _coerce(await original_aexecute(self, *args, **kwargs))

    setattr(execute, _RESULT_PATCH_MARKER, True)
    setattr(aexecute, _RESULT_PATCH_MARKER, True)
    FunctionCall.execute = execute
    FunctionCall.aexecute = aexecute
    return True

patch_literal_enum_schema()

Advertise Literal parameters to the model as string enums.

Agno 2.1.9's schema builder has no branch for :data:typing.Literal, so every such parameter falls through to the generic object case and reaches the model as {"type": "object", "properties": {}} -- no values, no default. Pydantic then rejects the call at runtime with "Input should be 'x' or 'y'", penalising the model for a constraint it was never shown, and a model that obeys the advertised schema sends a dict and fails the same way.

Returns whether the patch is active.

Source code in src/cs_copilot/agents/agno_compat.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def patch_literal_enum_schema() -> bool:
    """Advertise ``Literal`` parameters to the model as string enums.

    Agno 2.1.9's schema builder has no branch for :data:`typing.Literal`, so
    every such parameter falls through to the generic object case and reaches
    the model as ``{"type": "object", "properties": {}}`` -- no values, no
    default. Pydantic then rejects the call at runtime with "Input should be
    'x' or 'y'", penalising the model for a constraint it was never shown, and
    a model that obeys the advertised schema sends a dict and fails the same
    way.

    Returns whether the patch is active.
    """

    import typing

    from agno.utils import json_schema as agno_json_schema

    original = agno_json_schema.get_json_schema_for_arg
    if getattr(original, _LITERAL_PATCH_MARKER, False):
        return True
    try:
        installed = version("agno")
    except PackageNotFoundError:  # pragma: no cover - agno is a hard dependency
        return False
    if installed not in PATCHED_AGNO_VERSIONS:
        logger.warning(
            "Not applying the Literal enum schema fix for agno %s (validated for %s); "
            "check whether get_json_schema_for_arg still needs it.",
            installed,
            ", ".join(sorted(PATCHED_AGNO_VERSIONS)),
        )
        return False

    _JSON_TYPES = ((str, "string"), (bool, "boolean"), (int, "integer"), (float, "number"))

    def get_json_schema_for_arg(type_hint, *args, **kwargs):
        if typing.get_origin(type_hint) is typing.Literal:
            values = list(typing.get_args(type_hint))
            for python_type, json_type in _JSON_TYPES:
                # bool before int: bool is a subclass of int.
                if values and all(isinstance(value, python_type) for value in values):
                    return {"type": json_type, "enum": values}
        return original(type_hint, *args, **kwargs)

    setattr(get_json_schema_for_arg, _LITERAL_PATCH_MARKER, True)
    agno_json_schema.get_json_schema_for_arg = get_json_schema_for_arg
    return True

config

Configuration module for cs_copilot agents.

Agent personas and behavioral instructions live in descriptions.py and instructions.py respectively.

context

Explicit context budgets for bounded, auditable agent handoffs.

ContextBudget dataclass

Per-component token allocation for a specialist invocation.

Source code in src/cs_copilot/agents/context.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@dataclass(frozen=True)
class ContextBudget:
    """Per-component token allocation for a specialist invocation."""

    system_policy: int = 1_500
    procedure: int = 3_000
    run_summary: int = 2_000
    tool_schemas: int = 3_000
    recent_messages: int = 1_500

    def __post_init__(self) -> None:
        for name, value in self.as_dict().items():
            if value < 0:
                raise ValueError(f"context budget {name} cannot be negative")
        if self.total <= 0:
            raise ValueError("context budget total must be positive")

    @property
    def total(self) -> int:
        return sum(self.as_dict().values())

    def as_dict(self) -> dict[str, int]:
        return {
            "system_policy": self.system_policy,
            "procedure": self.procedure,
            "run_summary": self.run_summary,
            "tool_schemas": self.tool_schemas,
            "recent_messages": self.recent_messages,
        }

BoundedContext dataclass

Prompt sections and their measured size after explicit truncation.

Source code in src/cs_copilot/agents/context.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@dataclass(frozen=True)
class BoundedContext:
    """Prompt sections and their measured size after explicit truncation."""

    sections: Mapping[str, str]
    token_counts: Mapping[str, int]
    truncated: tuple[str, ...]

    @property
    def total_tokens(self) -> int:
        return sum(self.token_counts.values())

    def render(self) -> str:
        return "\n\n".join(
            f"## {name.replace('_', ' ').title()}\n{text}"
            for name, text in self.sections.items()
            if text
        )

ContextBuilder

Build bounded receiver context without forwarding a full conversation.

Source code in src/cs_copilot/agents/context.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class ContextBuilder:
    """Build bounded receiver context without forwarding a full conversation."""

    def __init__(
        self,
        budget: ContextBudget = DEFAULT_CONTEXT_BUDGET,
        *,
        count_tokens: Callable[[str], int] | None = None,
    ) -> None:
        self.budget = budget
        self._count_tokens = count_tokens or _estimate_tokens

    def build(
        self,
        handoff: HandoffEnvelope,
        *,
        system_policy: str,
        procedure: str,
        run_summary: str,
        tool_schemas: str,
        recent_messages: Iterable[str] = (),
    ) -> BoundedContext:
        """Fit each component independently and report every truncation."""
        handoff_text = _render_handoff(handoff)
        recent_text = _bounded_recent_messages(recent_messages)
        raw_sections = {
            "system_policy": system_policy,
            "procedure": f"{handoff_text}\n\nProcedure:\n{procedure}".strip(),
            "run_summary": run_summary,
            "tool_schemas": tool_schemas,
            "recent_messages": recent_text,
        }
        allocations = self.budget.as_dict()
        sections: dict[str, str] = {}
        token_counts: dict[str, int] = {}
        truncated: list[str] = []
        for name, text in raw_sections.items():
            fitted, was_truncated = _fit_text(text or "", allocations[name], self._count_tokens)
            sections[name] = fitted
            token_counts[name] = self._count_tokens(fitted)
            if was_truncated:
                truncated.append(name)
        return BoundedContext(sections, token_counts, tuple(truncated))
build(handoff, *, system_policy, procedure, run_summary, tool_schemas, recent_messages=())

Fit each component independently and report every truncation.

Source code in src/cs_copilot/agents/context.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def build(
    self,
    handoff: HandoffEnvelope,
    *,
    system_policy: str,
    procedure: str,
    run_summary: str,
    tool_schemas: str,
    recent_messages: Iterable[str] = (),
) -> BoundedContext:
    """Fit each component independently and report every truncation."""
    handoff_text = _render_handoff(handoff)
    recent_text = _bounded_recent_messages(recent_messages)
    raw_sections = {
        "system_policy": system_policy,
        "procedure": f"{handoff_text}\n\nProcedure:\n{procedure}".strip(),
        "run_summary": run_summary,
        "tool_schemas": tool_schemas,
        "recent_messages": recent_text,
    }
    allocations = self.budget.as_dict()
    sections: dict[str, str] = {}
    token_counts: dict[str, int] = {}
    truncated: list[str] = []
    for name, text in raw_sections.items():
        fitted, was_truncated = _fit_text(text or "", allocations[name], self._count_tokens)
        sections[name] = fitted
        token_counts[name] = self._count_tokens(fitted)
        if was_truncated:
            truncated.append(name)
    return BoundedContext(sections, token_counts, tuple(truncated))

contracts

Typed contracts for specialist roles and multi-agent handoffs.

These contracts deliberately carry task facts and artifact references only. Conversation histories, scratchpads, private reasoning, and chain-of-thought are not valid handoff fields.

ExecutionBudget dataclass

Bounded resources a receiving role may spend on one task.

Source code in src/cs_copilot/agents/contracts.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@dataclass(frozen=True)
class ExecutionBudget:
    """Bounded resources a receiving role may spend on one task."""

    max_tokens: int | None = None
    max_tool_calls: int | None = None
    timeout_seconds: float | None = None

    def __post_init__(self) -> None:
        for name, value in (
            ("max_tokens", self.max_tokens),
            ("max_tool_calls", self.max_tool_calls),
            ("timeout_seconds", self.timeout_seconds),
        ):
            if value is not None and value <= 0:
                raise ValueError(f"{name} must be positive when provided")

    def as_dict(self) -> dict[str, int | float | None]:
        return {
            "max_tokens": self.max_tokens,
            "max_tool_calls": self.max_tool_calls,
            "timeout_seconds": self.timeout_seconds,
        }

RolePolicy dataclass

Allowlist for one in-process Agno specialist role.

Source code in src/cs_copilot/agents/contracts.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@dataclass(frozen=True)
class RolePolicy:
    """Allowlist for one in-process Agno specialist role."""

    role: str
    profile: str
    allowed_toolkits: frozenset[str]
    allowed_functions: frozenset[str] = frozenset()

    def allows(self, tool: Any) -> bool:
        if callable(tool) and hasattr(tool, "__name__") and not hasattr(tool, "__dict__"):
            return tool.__name__ in self.allowed_functions
        if callable(tool) and getattr(tool, "__name__", None) in self.allowed_functions:
            return True
        return tool.__class__.__name__ in self.allowed_toolkits

    def as_dict(self) -> dict[str, Any]:
        return {
            "role": self.role,
            "profile": self.profile,
            "allowed_toolkits": sorted(self.allowed_toolkits),
            "allowed_functions": sorted(self.allowed_functions),
        }

validate_role_tools(policy, tools)

Reject a factory configuration that grants tools outside its role policy.

Source code in src/cs_copilot/agents/contracts.py 
101
102
103
104
105
106
107
108
def validate_role_tools(policy: RolePolicy, tools: Iterable[Any]) -> None:
    """Reject a factory configuration that grants tools outside its role policy."""
    denied = [_tool_identity(tool) for tool in tools if not policy.allows(tool)]
    if denied:
        raise ValueError(
            f"Role '{policy.role}' is not allowed to use configured tools: "
            + ", ".join(sorted(denied))
        )

record_handoff(runtime, envelope)

Record a handoff through a v2 RunContext when one is available.

Source code in src/cs_copilot/agents/contracts.py 
111
112
113
114
115
116
117
118
119
def record_handoff(runtime: Any, envelope: HandoffEnvelope) -> bool:
    """Record a handoff through a v2 ``RunContext`` when one is available."""
    if runtime is None:
        return False
    recorder = getattr(runtime, "record_handoff", None)
    if not callable(recorder):
        raise TypeError("workflow runtime does not expose record_handoff")
    recorder(envelope)
    return True

delegation

Agno coordinator guard for structured, bounded specialist delegation.

DelegationLimits dataclass

Hard limits applied to one coordinator run.

Source code in src/cs_copilot/agents/delegation.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@dataclass(frozen=True)
class DelegationLimits:
    """Hard limits applied to one coordinator run."""

    max_delegations_per_run: int = 12
    max_delegations_per_member: int = 5
    max_delegations_per_task: int = 2
    max_identical_handoffs: int = 1
    max_handoff_bytes: int = 32_000
    max_tokens: int = 12_000
    max_tool_calls: int = 24
    max_timeout_seconds: float = 1_800
    max_tracked_runs: int = 64

    def __post_init__(self) -> None:
        integer_limits = (
            "max_delegations_per_run",
            "max_delegations_per_member",
            "max_delegations_per_task",
            "max_identical_handoffs",
            "max_handoff_bytes",
            "max_tokens",
            "max_tool_calls",
            "max_tracked_runs",
        )
        for name in integer_limits:
            value = getattr(self, name)
            if isinstance(value, bool) or not isinstance(value, int):
                raise TypeError(f"delegation limit {name} must be an integer")
        if isinstance(self.max_timeout_seconds, bool) or not isinstance(
            self.max_timeout_seconds, (int, float)
        ):
            raise TypeError("delegation limit max_timeout_seconds must be numeric")
        if not math.isfinite(float(self.max_timeout_seconds)):
            raise ValueError("delegation limit max_timeout_seconds must be finite")
        for name, value in self.as_dict().items():
            if value <= 0:
                raise ValueError(f"delegation limit {name} must be positive")

    def as_dict(self) -> dict[str, int | float]:
        return {
            "max_delegations_per_run": self.max_delegations_per_run,
            "max_delegations_per_member": self.max_delegations_per_member,
            "max_delegations_per_task": self.max_delegations_per_task,
            "max_identical_handoffs": self.max_identical_handoffs,
            "max_handoff_bytes": self.max_handoff_bytes,
            "max_tokens": self.max_tokens,
            "max_tool_calls": self.max_tool_calls,
            "max_timeout_seconds": self.max_timeout_seconds,
            "max_tracked_runs": self.max_tracked_runs,
        }

StructuredDelegationGuard

Validate and canonicalize Agno's generated delegation tool calls.

The guard is installed as the generated delegation Function.pre_hook. Agno invokes synchronous pre-hooks in both its sync and async execution paths, so validation is identical without wrapping unrelated team tools.

Source code in src/cs_copilot/agents/delegation.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
class StructuredDelegationGuard:
    """Validate and canonicalize Agno's generated delegation tool calls.

    The guard is installed as the generated delegation ``Function.pre_hook``.
    Agno invokes synchronous pre-hooks in both its sync and async execution
    paths, so validation is identical without wrapping unrelated team tools.
    """

    def __init__(self, limits: DelegationLimits = DEFAULT_DELEGATION_LIMITS) -> None:
        self.limits = limits
        self._states: OrderedDict[str, _RunDelegationState] = OrderedDict()
        self._lock = Lock()

    def pre_hook(
        self,
        *,
        team: Any = None,
        session_state: Mapping[str, Any] | None = None,
        fc: Any = None,
    ) -> None:
        """Agno ``Function.pre_hook`` entry point."""
        self._pre_hook(team=team, session_state=session_state, fc=fc)

    def hook_for_run(self, coordinator_run_id: str | None) -> Any:
        """Bind counters to Agno's immutable run ID instead of mutable session state."""

        def validate_structured_delegation(
            *,
            team: Any = None,
            session_state: Mapping[str, Any] | None = None,
            fc: Any = None,
        ) -> None:
            self._pre_hook(
                team=team,
                session_state=session_state,
                fc=fc,
                coordinator_run_id=coordinator_run_id,
            )

        return validate_structured_delegation

    def _pre_hook(
        self,
        *,
        team: Any,
        session_state: Mapping[str, Any] | None,
        fc: Any,
        coordinator_run_id: str | None = None,
    ) -> None:
        function = getattr(fc, "function", None)
        function_name = getattr(function, "name", "")
        if function_name not in {DELEGATE_TOOL_NAME, DELEGATE_ALL_TOOL_NAME}:
            return
        arguments = getattr(fc, "arguments", None)
        dependencies = getattr(function, "_dependencies", None)
        self._guard_call(
            function_name=function_name,
            arguments=arguments,
            team=team,
            session_state=session_state,
            dependencies=dependencies,
            coordinator_run_id=coordinator_run_id,
        )

    def _guard_call(
        self,
        *,
        function_name: str,
        arguments: MutableMapping[str, Any],
        team: Any,
        session_state: Mapping[str, Any] | None,
        dependencies: Mapping[str, Any] | None,
        coordinator_run_id: str | None,
    ) -> None:
        if function_name == DELEGATE_ALL_TOOL_NAME:
            raise RetryAgentRun(
                "Broadcast delegation is disabled: create one role-scoped structured "
                "handoff per specialist."
            )

        try:
            envelope = self._validate_call(arguments, team)
            fingerprint = _handoff_fingerprint(envelope)
            run_key = coordinator_run_id or _coordinator_run_key(session_state, envelope)
            self._reserve(run_key, arguments["member_id"], envelope.task_id, fingerprint)
            try:
                runtime = _resolve_run_context(team, dependencies)
                envelope = _record_delegation(team, runtime, envelope)
            except Exception:
                self._release(
                    run_key,
                    arguments["member_id"],
                    envelope.task_id,
                    fingerprint,
                )
                raise
            _canonicalize_delegate_arguments(arguments, envelope)
        except RetryAgentRun:
            raise
        except Exception as exc:
            raise RetryAgentRun(
                "Delegation rejected by the structured handoff guard: "
                f"{exc}. Pass task_description as one JSON handoff object."
            ) from exc

    def _validate_call(
        self,
        arguments: MutableMapping[str, Any],
        team: Any,
    ) -> HandoffEnvelope:
        if not isinstance(arguments, MutableMapping):
            raise TypeError("delegation arguments must be a mapping")
        unexpected = sorted(set(arguments) - _ALLOWED_DELEGATE_ARGUMENTS)
        if unexpected:
            raise ValueError(f"unexpected delegation arguments: {', '.join(unexpected)}")

        member_id = str(arguments.get("member_id") or "").strip()
        if not member_id:
            raise ValueError("member_id is required")
        raw_description = arguments.get("task_description")
        if not isinstance(raw_description, str) or not raw_description.strip():
            raise TypeError("task_description must be a non-empty JSON string")
        if len(raw_description.encode("utf-8")) > self.limits.max_handoff_bytes:
            raise ValueError(f"handoff exceeds {self.limits.max_handoff_bytes} encoded bytes")

        try:
            payload = json.loads(
                raw_description,
                object_pairs_hook=_mapping_without_duplicate_keys,
                parse_constant=_reject_non_finite_json,
            )
        except json.JSONDecodeError as exc:
            raise ValueError("task_description is not valid JSON") from exc
        if not isinstance(payload, Mapping):
            raise TypeError("task_description JSON must contain an object")
        _reject_private_context(payload)

        missing = sorted(_REQUIRED_HANDOFF_FIELDS - set(payload))
        if missing:
            raise ValueError(f"handoff is missing required fields: {', '.join(missing)}")
        unexpected = sorted(set(payload) - _ALLOWED_HANDOFF_FIELDS)
        if unexpected:
            raise ValueError(f"handoff has unknown fields: {', '.join(unexpected)}")
        _validate_structured_fields(payload)
        self._validate_budget(payload["budget"])

        envelope = HandoffEnvelope.from_mapping(payload)
        selected_member = _selected_member(team, member_id)
        selected_role = _member_role(selected_member)
        if selected_role is None:
            raise ValueError(f"member '{member_id}' has no declared agentic role")
        if envelope.receiver_role != selected_role:
            raise ValueError(
                f"receiver_role '{envelope.receiver_role}' does not match member "
                f"'{member_id}' role '{selected_role}'"
            )
        allowed_senders = set(ROLE_POLICIES) | {COORDINATOR_ROLE}
        if envelope.sender_role not in allowed_senders:
            raise ValueError(f"unknown sender_role '{envelope.sender_role}'")
        if envelope.receiver_role not in ROLE_POLICIES:
            raise ValueError(f"unknown receiver_role '{envelope.receiver_role}'")
        _suppress_member_context(selected_member)
        return envelope

    def _validate_budget(self, value: Any) -> None:
        if not isinstance(value, Mapping):
            raise TypeError("handoff budget must be a mapping")
        missing = sorted(_REQUIRED_BUDGET_FIELDS - set(value))
        if missing:
            raise ValueError(f"handoff budget is missing: {', '.join(missing)}")
        unexpected = sorted(set(value) - _REQUIRED_BUDGET_FIELDS)
        if unexpected:
            raise ValueError(f"handoff budget has unknown fields: {', '.join(unexpected)}")
        budget = ExecutionBudget(
            max_tokens=_strict_int(value["max_tokens"], "max_tokens"),
            max_tool_calls=_strict_int(value["max_tool_calls"], "max_tool_calls"),
            timeout_seconds=_strict_number(value["timeout_seconds"], "timeout_seconds"),
        )
        if budget.max_tokens is not None and budget.max_tokens > self.limits.max_tokens:
            raise ValueError(f"max_tokens exceeds guard limit {self.limits.max_tokens}")
        if budget.max_tool_calls is not None and budget.max_tool_calls > self.limits.max_tool_calls:
            raise ValueError(f"max_tool_calls exceeds guard limit {self.limits.max_tool_calls}")
        if (
            budget.timeout_seconds is not None
            and budget.timeout_seconds > self.limits.max_timeout_seconds
        ):
            raise ValueError(
                "timeout_seconds exceeds guard limit " f"{self.limits.max_timeout_seconds:g}"
            )

    def _reserve(
        self,
        run_key: str,
        member_id: str,
        task_id: str,
        fingerprint: str,
    ) -> None:
        with self._lock:
            state = self._state_for(run_key)
            if state.total >= self.limits.max_delegations_per_run:
                raise ValueError(
                    "coordinator delegation limit reached for this run "
                    f"({self.limits.max_delegations_per_run})"
                )
            if state.per_member[member_id] >= self.limits.max_delegations_per_member:
                raise ValueError(
                    f"member '{member_id}' delegation limit reached "
                    f"({self.limits.max_delegations_per_member})"
                )
            task_key = (member_id, task_id)
            if state.per_task[task_key] >= self.limits.max_delegations_per_task:
                raise ValueError(
                    f"task '{task_id}' repeat limit reached "
                    f"({self.limits.max_delegations_per_task})"
                )
            if state.fingerprints[fingerprint] >= self.limits.max_identical_handoffs:
                raise ValueError(
                    "identical handoff already delegated; revise the task or context "
                    "instead of repeating the same call"
                )
            state.total += 1
            state.per_member[member_id] += 1
            state.per_task[task_key] += 1
            state.fingerprints[fingerprint] += 1

    def _release(
        self,
        run_key: str,
        member_id: str,
        task_id: str,
        fingerprint: str,
    ) -> None:
        with self._lock:
            state = self._states.get(run_key)
            if state is None:
                return
            state.total = max(0, state.total - 1)
            _decrement(state.per_member, member_id)
            _decrement(state.per_task, (member_id, task_id))
            _decrement(state.fingerprints, fingerprint)

    def _state_for(self, run_key: str) -> _RunDelegationState:
        state = self._states.get(run_key)
        if state is not None:
            self._states.move_to_end(run_key)
            return state
        state = _RunDelegationState()
        self._states[run_key] = state
        while len(self._states) > self.limits.max_tracked_runs:
            self._states.popitem(last=False)
        return state
pre_hook(*, team=None, session_state=None, fc=None)

Agno Function.pre_hook entry point.

Source code in src/cs_copilot/agents/delegation.py 
167
168
169
170
171
172
173
174
175
def pre_hook(
    self,
    *,
    team: Any = None,
    session_state: Mapping[str, Any] | None = None,
    fc: Any = None,
) -> None:
    """Agno ``Function.pre_hook`` entry point."""
    self._pre_hook(team=team, session_state=session_state, fc=fc)
hook_for_run(coordinator_run_id)

Bind counters to Agno's immutable run ID instead of mutable session state.

Source code in src/cs_copilot/agents/delegation.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def hook_for_run(self, coordinator_run_id: str | None) -> Any:
    """Bind counters to Agno's immutable run ID instead of mutable session state."""

    def validate_structured_delegation(
        *,
        team: Any = None,
        session_state: Mapping[str, Any] | None = None,
        fc: Any = None,
    ) -> None:
        self._pre_hook(
            team=team,
            session_state=session_state,
            fc=fc,
            coordinator_run_id=coordinator_run_id,
        )

    return validate_structured_delegation

StructuredHandoffTeam

Bases: Team

Team variant that keeps implicit coordinator context out of member prompts.

Agno 2.1.9 has one add_history_to_context flag for both the coordinator and delegated members. The only narrow seam that can preserve coordinator history while disabling member history is the generated delegate-function builder. This override changes only those member-call flags.

Source code in src/cs_copilot/agents/delegation.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
class StructuredHandoffTeam(Team):
    """Team variant that keeps implicit coordinator context out of member prompts.

    Agno 2.1.9 has one ``add_history_to_context`` flag for both the coordinator
    and delegated members. The only narrow seam that can preserve coordinator
    history while disabling member history is the generated delegate-function
    builder. This override changes only those member-call flags.
    """

    def __init__(
        self,
        *args: Any,
        run_context: Any = None,
        delegation_guard: StructuredDelegationGuard | None = None,
        **kwargs: Any,
    ) -> None:
        super().__init__(*args, **kwargs)
        collisions = sorted(
            _RESERVED_DELEGATE_TOOL_NAMES.intersection(_configured_tool_names(self.tools or ()))
        )
        if collisions:
            raise ValueError(
                "configured team tools cannot shadow Agno delegation tools: "
                + ", ".join(collisions)
            )
        for member in self.members:
            _suppress_member_context(member)
        self.run_context = run_context
        self.delegation_guard = delegation_guard or StructuredDelegationGuard()

    def _get_delegate_task_function(self, *args: Any, **kwargs: Any) -> Any:
        run_response = kwargs.get("run_response")
        if run_response is None and args:
            run_response = args[0]
        coordinator_run_id = getattr(run_response, "run_id", None)
        kwargs["add_history_to_context"] = False
        kwargs["add_session_state_to_context"] = False
        kwargs["add_dependencies_to_context"] = False
        delegate_function = super()._get_delegate_task_function(*args, **kwargs)
        delegate_function.pre_hook = self.delegation_guard.hook_for_run(
            str(coordinator_run_id) if coordinator_run_id else None
        )
        delegate_function.description = (
            "Delegate one bounded task to one specialist. task_description must be a "
            "JSON-encoded v2 HandoffEnvelope; unstructured, private-history, oversized, "
            "or repeated handoffs are rejected."
        )
        properties = (delegate_function.parameters or {}).get("properties", {})
        if "task_description" in properties:
            properties["task_description"][
                "description"
            ] = "A JSON string containing the complete structured handoff contract."
        if "expected_output" in properties:
            properties["expected_output"][
                "description"
            ] = "Optional placeholder; the guard derives this value from the handoff."
        return delegate_function

descriptions

Agent persona descriptions for cs_copilot.

These strings answer who each agent is — the identity/role that Agno injects into the system message via Agent(description=...). They are deliberately kept separate from the behavioral rules in instructions.py (Agno's instructions=) and from the multi-tool procedures in the skill/workflow catalogs. Change a persona here; change how it behaves in instructions.py; change what it does step by step in skills/ or workflow_catalog/.

execution_binding

Route the in-process Agno team's toolkit calls through the execution kernel.

:func:attach_execution wraps every toolkit function of a team (coordinator toolkits and every member's toolkits and bare-function tools) with a signature-preserving entrypoint. Each call is then recorded in the chat's durable workflow run like an MCP tool call: same event protocol, same tool identity (via :mod:cs_copilot.agents.execution_contracts), same artifact registration.

The mode comes from CS_COPILOT_AGNO_EXECUTION (default enforce):

  • off: no wrapping; behaviour and storage are exactly as before.
  • observe: calls run unchanged; events, artifact registration, and an audit of what enforcement would rewrite or deny are recorded (:func:cs_copilot.execution.runner.observe_sync).
  • enforce (default): the full kernel (:func:cs_copilot.execution.runner.execute_sync): output paths are rewritten into the chat's run, writes are confined and create-only, declared file inputs must be registered artifacts, every file a call writes is registered, and a call's files are rolled back if it fails. Denials reach the model as tool errors.

Agno hooks are deliberately not used: Team.tool_hooks would also wrap the async-generator delegation tool, hooks cannot rewrite arguments, and they do not know which toolkit a call belongs to.

ExecutionBinding dataclass

Per-team routing state: the chat's run, the mode, and task attribution.

Source code in src/cs_copilot/agents/execution_binding.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
@dataclass
class ExecutionBinding:
    """Per-team routing state: the chat's run, the mode, and task attribution."""

    mode: ExecutionMode
    team: Any
    run_context: Any = None
    active_tasks: dict[str, str] = field(default_factory=dict)
    turn_tasks: list[str] = field(default_factory=list)
    problems: list[str] = field(default_factory=list)
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False)
    _in_flight: int = 0
    _idle: threading.Condition = field(init=False, repr=False)

    def __post_init__(self) -> None:
        self._idle = threading.Condition(self._lock)

    @property
    def enabled(self) -> bool:
        return self.mode is not ExecutionMode.OFF and self.run_context is not None

    def note_task_started(self, role: str, task_id: str) -> None:
        with self._lock:
            self.active_tasks[role] = task_id
            if task_id not in self.turn_tasks:
                self.turn_tasks.append(task_id)

    def take_turn_tasks(self) -> list[str]:
        with self._lock:
            tasks, self.turn_tasks = self.turn_tasks, []
            self.active_tasks.clear()
            return tasks

    def note_problem(self, message: str) -> None:
        with self._lock:
            self.problems = [*self.problems[-49:], message]

    def take_problems(self) -> list[str]:
        with self._lock:
            problems, self.problems = self.problems, []
            return problems

    def in_flight(self) -> int:
        with self._lock:
            return self._in_flight

    def wait_idle(self, timeout: float | None = None) -> bool:
        """Block until no bound call is running (used before cancelling a run)."""

        with self._idle:
            return self._idle.wait_for(lambda: self._in_flight == 0, timeout=timeout)

    def invoke(
        self,
        function: Callable[..., Any],
        owner: Any,
        args: tuple[Any, ...],
        kwargs: dict[str, Any],
        *,
        context: Mapping[str, Any] | None = None,
    ) -> Any:
        """Run one Agno tool call, recording it through the kernel when enabled.

        ``kwargs`` are what ``function`` receives. ``context`` holds Agno's
        ``agent`` / ``team`` / ``session_state`` for functions that do not
        declare them; they inform the recording but are never passed on.
        """

        if not self.enabled or _IN_BOUND_CALL.get():
            return function(*args, **kwargs)
        token = _IN_BOUND_CALL.set(True)
        with self._lock:
            self._in_flight += 1
        try:
            if self.mode is ExecutionMode.ENFORCE:
                return self._enforce(function, owner, args, kwargs, dict(context or {}))
            return self._observe(function, owner, args, kwargs, dict(context or {}))
        finally:
            _IN_BOUND_CALL.reset(token)
            with self._idle:
                self._in_flight -= 1
                self._idle.notify_all()

    def _observe(
        self,
        function: Callable[..., Any],
        owner: Any,
        args: tuple[Any, ...],
        kwargs: dict[str, Any],
        context: dict[str, Any],
    ) -> Any:
        run_context = self.run_context
        try:
            contract, _source = resolve_contract(function, owner=owner)
            known = {**context, **{key: kwargs[key] for key in _WRAPPER_INJECTED if key in kwargs}}
            state = known.get("session_state")
            if not isinstance(state, dict):
                state = self.team_state()
            self._ensure_bound(state)
            ctx = BasicExecutionContext(
                execution_runtime=AGNO_RUNTIME,
                session_state=state,
                run_context=run_context,
            )
            scope = self._scope(known.get("agent"))
            public = {key: value for key, value in kwargs.items() if key not in _AGNO_INJECTED}
            injected = {key: value for key, value in kwargs.items() if key in _AGNO_INJECTED}
        except Exception as exc:  # noqa: BLE001 - never block the call on bookkeeping
            logger.warning("Could not prepare kernel bookkeeping for %r: %s", function, exc)
            self.note_problem(f"prepare {getattr(function, '__name__', function)}: {exc}")
            return function(*args, **kwargs)

        def call(arguments: dict[str, Any]) -> Any:
            return function(*args, **arguments, **injected)

        with S3.scoped_session_prefix(f"sessions/{run_context.run.session_id}"):
            return observe_sync(
                contract,
                ctx,
                public,
                invoke=call,
                scope=scope,
                extra={"owner": PROCESS_OWNER, "execution_mode": self.mode.value},
            )

    def _enforce(
        self,
        function: Callable[..., Any],
        owner: Any,
        args: tuple[Any, ...],
        kwargs: dict[str, Any],
        context: dict[str, Any],
    ) -> Any:
        run_context = self.run_context
        contract, _source = resolve_contract(function, owner=owner)
        contract = _in_process_contract(contract, function)
        known = {**context, **{key: kwargs[key] for key in _WRAPPER_INJECTED if key in kwargs}}
        state = known.get("session_state")
        if not isinstance(state, dict):
            state = self.team_state()
        self._ensure_bound(state)
        ctx = BasicExecutionContext(
            execution_runtime=AGNO_RUNTIME,
            session_state=state,
            run_context=run_context,
        )
        public = {key: value for key, value in kwargs.items() if key not in _AGNO_INJECTED}
        injected = {key: value for key, value in kwargs.items() if key in _AGNO_INJECTED}

        def inject(call_kwargs: dict[str, Any], session_view: dict[str, Any] | None) -> None:
            call_kwargs.update(injected)
            if session_view is not None and "session_state" in injected:
                call_kwargs["session_state"] = session_view

        def call(call_kwargs: dict[str, Any]) -> Any:
            return function(*args, **call_kwargs)

        with S3.scoped_session_prefix(f"sessions/{run_context.run.session_id}"):
            outcome = execute_sync(
                contract,
                ctx,
                public,
                invoke=call,
                inject=inject,
                publication_policy="all_published",
                verify_artifacts_on_lock=False,
                scope=self._scope(known.get("agent")),
                extra={"owner": PROCESS_OWNER, "execution_mode": self.mode.value},
            )
        if outcome.ok:
            return outcome.value
        if outcome.error is not None:
            raise outcome.error
        raise ToolExecutionError(str((outcome.envelope.get("error") or {}).get("message")))

    def team_state(self) -> dict[str, Any]:
        state = getattr(self.team, "session_state", None)
        if not isinstance(state, dict):
            state = {}
            self.team.session_state = state
        return state

    def _ensure_bound(self, state: dict[str, Any]) -> None:
        run = self.run_context.run
        output_context = state.get(OUTPUT_CONTEXT_KEY)
        if (
            isinstance(output_context, Mapping)
            and output_context.get("run_id") == run.run_id
            and output_context.get("session_id") == run.session_id
        ):
            return
        self.run_context.bind_session_state(state)
        team_state = self.team_state()
        if team_state is not state:
            self.run_context.bind_session_state(team_state)
        self.note_problem("re-bound a session_state whose output_context did not name the run")

    def _scope(self, agent: Any) -> _InvocationScope:
        run = self.run_context.run
        role = getattr(agent, "agentic_role", None) if agent is not None else None
        if role is None:
            role = capabilities.COORDINATOR_ROLE
        policy = ROLE_POLICIES.get(role)
        with self._lock:
            task_id = self.active_tasks.get(role)
        task = run.tasks.get(task_id) if task_id else None
        return _InvocationScope(
            run_id=str(run.run_id),
            task_id=task_id if task is not None else None,
            role=role,
            profile=policy.profile if policy is not None else None,
            task_attempt=int(task.attempts) if task is not None else None,
        )
wait_idle(timeout=None)

Block until no bound call is running (used before cancelling a run).

Source code in src/cs_copilot/agents/execution_binding.py 
143
144
145
146
147
def wait_idle(self, timeout: float | None = None) -> bool:
    """Block until no bound call is running (used before cancelling a run)."""

    with self._idle:
        return self._idle.wait_for(lambda: self._in_flight == 0, timeout=timeout)
invoke(function, owner, args, kwargs, *, context=None)

Run one Agno tool call, recording it through the kernel when enabled.

kwargs are what function receives. context holds Agno's agent / team / session_state for functions that do not declare them; they inform the recording but are never passed on.

Source code in src/cs_copilot/agents/execution_binding.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def invoke(
    self,
    function: Callable[..., Any],
    owner: Any,
    args: tuple[Any, ...],
    kwargs: dict[str, Any],
    *,
    context: Mapping[str, Any] | None = None,
) -> Any:
    """Run one Agno tool call, recording it through the kernel when enabled.

    ``kwargs`` are what ``function`` receives. ``context`` holds Agno's
    ``agent`` / ``team`` / ``session_state`` for functions that do not
    declare them; they inform the recording but are never passed on.
    """

    if not self.enabled or _IN_BOUND_CALL.get():
        return function(*args, **kwargs)
    token = _IN_BOUND_CALL.set(True)
    with self._lock:
        self._in_flight += 1
    try:
        if self.mode is ExecutionMode.ENFORCE:
            return self._enforce(function, owner, args, kwargs, dict(context or {}))
        return self._observe(function, owner, args, kwargs, dict(context or {}))
    finally:
        _IN_BOUND_CALL.reset(token)
        with self._idle:
            self._in_flight -= 1
            self._idle.notify_all()

execution_mode_from_env(value=None)

Resolve an execution mode from value or CS_COPILOT_AGNO_EXECUTION.

Source code in src/cs_copilot/agents/execution_binding.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def execution_mode_from_env(value: str | ExecutionMode | None = None) -> ExecutionMode:
    """Resolve an execution mode from ``value`` or ``CS_COPILOT_AGNO_EXECUTION``."""

    if isinstance(value, ExecutionMode):
        return value
    raw = os.getenv(EXECUTION_MODE_ENV) if value is None else value
    if raw is None or not str(raw).strip():
        return DEFAULT_EXECUTION_MODE
    try:
        return ExecutionMode(str(raw).strip().lower())
    except ValueError as exc:
        raise ValueError(
            f"{EXECUTION_MODE_ENV} must be one of off, observe, enforce; got {raw!r}"
        ) from exc

attach_execution(team, *, run_context, mode=None)

Route team's tool calls through the kernel for run_context.

Idempotent: re-attaching updates the run and mode of the existing binding. Must run before the team's first run, because Agno processes tool entrypoints when it first builds the model's tool list.

Source code in src/cs_copilot/agents/execution_binding.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
def attach_execution(
    team: Any,
    *,
    run_context: Any,
    mode: ExecutionMode | str | None = None,
) -> ExecutionBinding | None:
    """Route ``team``'s tool calls through the kernel for ``run_context``.

    Idempotent: re-attaching updates the run and mode of the existing binding.
    Must run before the team's first run, because Agno processes tool
    entrypoints when it first builds the model's tool list.
    """

    selected = execution_mode_from_env(mode)
    run = getattr(run_context, "run", None)
    if selected is ExecutionMode.ENFORCE and run is not None and not _is_ad_hoc(run):
        raise ValueError(
            "enforce mode supports ad-hoc chat runs only; catalog workflow runs "
            "need MCP task scope"
        )
    existing: ExecutionBinding | None = getattr(team, BINDING_ATTRIBUTE, None)
    if selected is ExecutionMode.OFF:
        if existing is not None:
            existing.mode = selected
        return existing
    if existing is not None:
        existing.mode = selected
        existing.run_context = run_context
        return existing
    binding = ExecutionBinding(mode=selected, team=team, run_context=run_context)
    setattr(team, BINDING_ATTRIBUTE, binding)
    for entity in _team_entities(team):
        _wrap_tools(entity, binding)
        _guard_agentic_state(entity)
    return binding

execution_contracts

Map in-process Agno tool functions to execution-kernel contracts.

Every MCP tool spec names the in-process implementations of its operation in ToolSpec.agno_bindings (module:Class.method or module:function). This module indexes those bindings by the function object they resolve to, so an Agno toolkit call is recorded under the same tool identity and execution contract as its MCP twin. Functions without an MCP twin get a conservative synthesized contract named agno.<Owner>.<function>.

The MCP tool catalog (cs_copilot.mcp.tools_registry) is imported lazily; the MCP server module and the mcp SDK never are.

function_key(function)

Identify a callable by its defining module and qualified name.

Source code in src/cs_copilot/agents/execution_contracts.py 
34
35
36
37
38
39
def function_key(function: Callable[..., Any]) -> FunctionKey:
    """Identify a callable by its defining module and qualified name."""

    target = inspect.unwrap(getattr(function, "__func__", function))
    target = getattr(target, "__func__", target)
    return str(target.__module__), str(target.__qualname__)

binding_target(binding)

Import the function a module:Class.method / module:function names.

Source code in src/cs_copilot/agents/execution_contracts.py 
42
43
44
45
46
47
48
49
def binding_target(binding: str) -> Callable[..., Any]:
    """Import the function a ``module:Class.method`` / ``module:function`` names."""

    module_name, _, attribute = binding.partition(":")
    target: Any = importlib.import_module(module_name)
    for part in attribute.split("."):
        target = getattr(target, part)
    return target

contract_index()

Return (building once) the function-to-contract index of the MCP catalog.

Source code in src/cs_copilot/agents/execution_contracts.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def contract_index() -> Mapping[FunctionKey, ToolSpec]:
    """Return (building once) the function-to-contract index of the MCP catalog."""

    global _INDEX
    with _LOCK:
        if _INDEX is None:
            from cs_copilot.mcp.tools_registry import all_specs

            index: dict[FunctionKey, ToolSpec] = {}
            for spec in all_specs():
                for binding in spec.agno_bindings:
                    key = function_key(binding_target(binding))
                    claimed = index.get(key)
                    if claimed is not None and claimed.mcp_name != spec.mcp_name:
                        raise ValueError(
                            f"{'.'.join(key)} is bound to both {claimed.mcp_name!r} "
                            f"and {spec.mcp_name!r}"
                        )
                    index[key] = spec
            _INDEX = index
        return _INDEX

resolve_contract(function, *, owner=None)

Return (contract, source) for an Agno tool function.

source is "spec" for a function bound to an MCP tool spec and "default" for a synthesized contract. owner is the toolkit instance (or None for a bare function tool).

Source code in src/cs_copilot/agents/execution_contracts.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def resolve_contract(
    function: Callable[..., Any],
    *,
    owner: Any = None,
) -> tuple[ToolSpec, str]:
    """Return ``(contract, source)`` for an Agno tool function.

    ``source`` is ``"spec"`` for a function bound to an MCP tool spec and
    ``"default"`` for a synthesized contract. ``owner`` is the toolkit instance
    (or ``None`` for a bare function tool).
    """

    key = function_key(function)
    spec = contract_index().get(key)
    if spec is not None:
        return spec, "spec"
    with _LOCK:
        default = _DEFAULTS.get(key)
        if default is None:
            default = _default_contract(function, owner)
            _DEFAULTS[key] = default
    return default, "default"

reset_contract_index()

Forget the cached index (tests and catalog reloads).

Source code in src/cs_copilot/agents/execution_contracts.py 
122
123
124
125
126
127
128
def reset_contract_index() -> None:
    """Forget the cached index (tests and catalog reloads)."""

    global _INDEX
    with _LOCK:
        _INDEX = None
        _DEFAULTS.clear()

factories

Agent factory classes for creating specialized cs_copilot agents. Contains the base factory class and all specialized factory implementations.

AgentConfig dataclass

Configuration for creating an agent.

Source code in src/cs_copilot/agents/factories.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@dataclass
class AgentConfig:
    """Configuration for creating an agent."""

    name: str
    description: str
    tools: List[Any] = field(default_factory=list)
    instructions: List[str] = field(default_factory=list)
    session_state: Dict[str, Any] = field(default_factory=dict)
    role_policy: Optional[RolePolicy] = None

    def validate(self) -> None:
        """Validate the agent configuration."""
        if not self.name:
            raise ValueError("Agent name cannot be empty")
        if not self.description:
            raise ValueError("Agent description cannot be empty")
        if not isinstance(self.tools, list):
            raise TypeError("Tools must be a list")
        if not isinstance(self.instructions, list):
            raise TypeError("Instructions must be a list")
        if self.role_policy is not None:
            validate_role_tools(self.role_policy, self.tools)
validate()

Validate the agent configuration.

Source code in src/cs_copilot/agents/factories.py 
82
83
84
85
86
87
88
89
90
91
92
93
def validate(self) -> None:
    """Validate the agent configuration."""
    if not self.name:
        raise ValueError("Agent name cannot be empty")
    if not self.description:
        raise ValueError("Agent description cannot be empty")
    if not isinstance(self.tools, list):
        raise TypeError("Tools must be a list")
    if not isinstance(self.instructions, list):
        raise TypeError("Instructions must be a list")
    if self.role_policy is not None:
        validate_role_tools(self.role_policy, self.tools)

AgentCreationError

Bases: Exception

Exception raised when agent creation fails.

Source code in src/cs_copilot/agents/factories.py 
96
97
98
99
class AgentCreationError(Exception):
    """Exception raised when agent creation fails."""

    pass

BaseAgentFactory

Bases: ABC

Base class for creating agents with common configuration and error handling.

Source code in src/cs_copilot/agents/factories.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
class BaseAgentFactory(ABC):
    """Base class for creating agents with common configuration and error handling."""

    def __init__(self, logger: Optional[logging.Logger] = None):
        self.logger = logger or logging.getLogger(__name__)

    @abstractmethod
    def get_agent_config(self) -> AgentConfig:
        """Return the configuration for this agent type."""
        pass

    def create_agent(
        self,
        model: Model,
        markdown: bool = True,
        debug_mode: bool = False,
        enable_mlflow_tracking: bool = True,
        **kwargs: Any,
    ) -> Agent:
        """Create an agent with error handling and validation.

        Args:
            model: Model to use for the agent
            markdown: Whether to enable markdown formatting
            debug_mode: Whether to enable debug mode
            enable_mlflow_tracking: Whether to enable MLflow tracking for this agent
            **kwargs: Additional keyword arguments for agent creation

        Returns:
            Created agent instance
        """
        try:
            config = self.get_agent_config()
            agent_type = getattr(self.__class__, "agent_type", None)
            declared_policy = ROLE_POLICIES.get(agent_type)
            if config.role_policy is None:
                config.role_policy = declared_policy
            elif declared_policy is not None and config.role_policy != declared_policy:
                raise ValueError(
                    f"Factory role policy for '{agent_type}' differs from the canonical allowlist"
                )
            config.validate()
            provided_session_state = kwargs.pop("session_state", None)

            # Log agent creation
            self.logger.info(f"Creating agent: {config.name}")

            # Create agent with common parameters
            agent_kwargs = {
                "model": model,
                "name": config.name,
                "description": config.description,
                "tools": config.tools,
                "markdown": markdown,
                "debug_mode": debug_mode,
                "enable_agentic_state": True,
                "add_session_state_to_context": True,
            }

            # Add optional parameters if they exist
            if config.instructions:
                agent_kwargs["instructions"] = config.instructions
            if provided_session_state is not None:
                if config.session_state:
                    _merge_session_state_defaults(provided_session_state, config.session_state)
                agent_kwargs["session_state"] = provided_session_state
            elif config.session_state:
                agent_kwargs["session_state"] = config.session_state

            # Add any additional kwargs passed in
            agent_kwargs.update(kwargs)

            agent = Agent(**agent_kwargs)
            if config.role_policy is not None:
                # Keep the enforced policy inspectable without placing it in the
                # shared session state (where one member could overwrite another).
                agent.role_policy = config.role_policy

            # Wrap agent methods with MLflow tracking if enabled
            if enable_mlflow_tracking:
                agent = self._wrap_agent_with_tracking(agent, config)

            self.logger.info(f"Successfully created agent: {config.name}")
            return agent

        except Exception as e:
            self.logger.error(
                f"Failed to create agent {config.name if 'config' in locals() else 'unknown'}: {str(e)}"
            )
            raise AgentCreationError(f"Failed to create agent: {str(e)}") from e

    def _wrap_agent_with_tracking(self, agent: Agent, config: AgentConfig) -> Agent:
        """Wrap agent execution methods with MLflow tracking.

        Args:
            agent: Agent instance to wrap
            config: Agent configuration

        Returns:
            Agent with wrapped methods
        """
        try:
            from cs_copilot.tracking import get_tracker
            from cs_copilot.tracking.utils import build_prompt_signature

            tracker = get_tracker()

            if not tracker.is_enabled():
                return agent

            # Get the agent type from the factory
            agent_type = getattr(self.__class__, "agent_type", None)

            def build_prompt_template() -> Optional[str]:
                sections = []
                if config.description:
                    sections.append(str(config.description).strip())
                if config.instructions:
                    normalized = [
                        str(item).strip() for item in config.instructions if item is not None
                    ]
                    instructions_text = "\n".join(normalized).strip()
                    if instructions_text:
                        sections.append(instructions_text)
                template = "\n\n".join([section for section in sections if section])
                return template.strip() if template else None

            def build_prompt_name() -> str:
                base_name = agent_type or config.name
                safe_name = str(base_name).replace(" ", "_").lower()
                return f"cs_copilot.{safe_name}"

            prompt_template = build_prompt_template()
            prompt_signature = build_prompt_signature(prompt_template)
            prompt_registry_name = build_prompt_name()

            def register_prompt_in_registry():
                if not prompt_template:
                    return
                commit_message = None
                if prompt_signature:
                    commit_message = f"cs_copilot auto update ({prompt_signature.version})"
                prompt_obj = tracker.register_prompt_version(
                    name=prompt_registry_name,
                    template=prompt_template,
                    commit_message=commit_message,
                    tags={
                        "agent_name": agent.name,
                        "agent_type": agent_type or "unknown",
                        "component": "cs_copilot",
                    },
                )
                if prompt_obj:
                    version = getattr(prompt_obj, "version", None)
                    tracker.log_params(
                        {
                            "prompt_registry_name": prompt_registry_name,
                            "prompt_registry_version": str(version) if version is not None else "",
                            "prompt_registry_uri": (
                                f"prompts:/{prompt_registry_name}/{version}"
                                if version is not None
                                else ""
                            ),
                        }
                    )

            # Wrap run() method
            original_run = agent.run

            def tracked_run(*args, **kwargs):
                # Extract prompt from args
                prompt = args[0] if args else kwargs.get("message", "")

                with tracker.track_agent_run(
                    agent_name=agent.name, prompt=str(prompt), agent_type=agent_type
                ):
                    # Log agent configuration
                    tracker.log_params(
                        {
                            "agent_name": agent.name,
                            "agent_type": agent_type or "unknown",
                            "num_tools": len(config.tools),
                            "tools": ",".join([t.__class__.__name__ for t in config.tools]),
                        }
                    )
                    register_prompt_in_registry()

                    result = original_run(*args, **kwargs)

                    # Log result metrics if available
                    if hasattr(result, "content") and result.content:
                        from cs_copilot.tracking.utils import count_tokens

                        tracker.log_metrics(
                            {"output_tokens_estimate": float(count_tokens(result.content))}
                        )

                    return result

            agent.run = tracked_run

            # Wrap arun() method (async version)
            original_arun = agent.arun

            async def tracked_arun(*args, **kwargs):
                # Extract prompt from args
                prompt = args[0] if args else kwargs.get("message", "")

                with tracker.track_agent_run(
                    agent_name=agent.name, prompt=str(prompt), agent_type=agent_type
                ):
                    # Log agent configuration
                    tracker.log_params(
                        {
                            "agent_name": agent.name,
                            "agent_type": agent_type or "unknown",
                            "num_tools": len(config.tools),
                            "tools": ",".join([t.__class__.__name__ for t in config.tools]),
                        }
                    )
                    register_prompt_in_registry()

                    result = await original_arun(*args, **kwargs)

                    # Log result metrics if available
                    if hasattr(result, "content") and result.content:
                        from cs_copilot.tracking.utils import count_tokens

                        tracker.log_metrics(
                            {"output_tokens_estimate": float(count_tokens(result.content))}
                        )

                    return result

            agent.arun = tracked_arun

            self.logger.debug(f"MLflow tracking enabled for agent: {agent.name}")

        except ImportError:
            self.logger.warning(
                "MLflow tracking module not available. Agent will run without tracking."
            )
        except Exception as e:
            self.logger.warning(f"Failed to enable MLflow tracking for agent: {e}")

        return agent
get_agent_config() abstractmethod

Return the configuration for this agent type.

Source code in src/cs_copilot/agents/factories.py 
118
119
120
121
@abstractmethod
def get_agent_config(self) -> AgentConfig:
    """Return the configuration for this agent type."""
    pass
create_agent(model, markdown=True, debug_mode=False, enable_mlflow_tracking=True, **kwargs)

Create an agent with error handling and validation.

Parameters:

Name Type Description Default
model Model

Model to use for the agent

 required
markdown bool

Whether to enable markdown formatting

 True
debug_mode bool

Whether to enable debug mode

 False
enable_mlflow_tracking bool

Whether to enable MLflow tracking for this agent

 True
**kwargs Any

Additional keyword arguments for agent creation

 {}

Returns:

Type Description
Agent

Created agent instance
Source code in src/cs_copilot/agents/factories.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def create_agent(
    self,
    model: Model,
    markdown: bool = True,
    debug_mode: bool = False,
    enable_mlflow_tracking: bool = True,
    **kwargs: Any,
) -> Agent:
    """Create an agent with error handling and validation.

    Args:
        model: Model to use for the agent
        markdown: Whether to enable markdown formatting
        debug_mode: Whether to enable debug mode
        enable_mlflow_tracking: Whether to enable MLflow tracking for this agent
        **kwargs: Additional keyword arguments for agent creation

    Returns:
        Created agent instance
    """
    try:
        config = self.get_agent_config()
        agent_type = getattr(self.__class__, "agent_type", None)
        declared_policy = ROLE_POLICIES.get(agent_type)
        if config.role_policy is None:
            config.role_policy = declared_policy
        elif declared_policy is not None and config.role_policy != declared_policy:
            raise ValueError(
                f"Factory role policy for '{agent_type}' differs from the canonical allowlist"
            )
        config.validate()
        provided_session_state = kwargs.pop("session_state", None)

        # Log agent creation
        self.logger.info(f"Creating agent: {config.name}")

        # Create agent with common parameters
        agent_kwargs = {
            "model": model,
            "name": config.name,
            "description": config.description,
            "tools": config.tools,
            "markdown": markdown,
            "debug_mode": debug_mode,
            "enable_agentic_state": True,
            "add_session_state_to_context": True,
        }

        # Add optional parameters if they exist
        if config.instructions:
            agent_kwargs["instructions"] = config.instructions
        if provided_session_state is not None:
            if config.session_state:
                _merge_session_state_defaults(provided_session_state, config.session_state)
            agent_kwargs["session_state"] = provided_session_state
        elif config.session_state:
            agent_kwargs["session_state"] = config.session_state

        # Add any additional kwargs passed in
        agent_kwargs.update(kwargs)

        agent = Agent(**agent_kwargs)
        if config.role_policy is not None:
            # Keep the enforced policy inspectable without placing it in the
            # shared session state (where one member could overwrite another).
            agent.role_policy = config.role_policy

        # Wrap agent methods with MLflow tracking if enabled
        if enable_mlflow_tracking:
            agent = self._wrap_agent_with_tracking(agent, config)

        self.logger.info(f"Successfully created agent: {config.name}")
        return agent

    except Exception as e:
        self.logger.error(
            f"Failed to create agent {config.name if 'config' in locals() else 'unknown'}: {str(e)}"
        )
        raise AgentCreationError(f"Failed to create agent: {str(e)}") from e

ChEMBLDownloaderFactory

Bases: BaseAgentFactory

Factory for creating ChemBL downloader agents.

Source code in src/cs_copilot/agents/factories.py 
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
class ChEMBLDownloaderFactory(BaseAgentFactory):
    """Factory for creating ChemBL downloader agents."""

    agent_type = "chembl_downloader"

    def get_agent_config(self) -> AgentConfig:
        return AgentConfig(
            name="chembl_agent",
            description=CHEMBL_DESCRIPTION,
            tools=[
                ChemblToolkit(),
                PointerPandasTools(),
                SessionMemoryToolkit(),
                SkillToolkit(),
            ],
            instructions=CHEMBL_INSTRUCTIONS,
            session_state={
                "data_file_paths": {
                    "dataset_path": None,  # Backward-compatible alias for clean_dataset_path.
                    "raw_dataset_path": None,
                    "clean_dataset_path": None,
                    "filtered_dataset_path": None,
                    "descriptor_parquet_path": None,
                    "standardization_report_path": None,
                }
            },
        )

ChemoinformaticianFactory

Bases: BaseAgentFactory

Factory for creating comprehensive chemoinformatics analysis agents.

This agent is a versatile chemoinformatician capable of: - Chemotype Analysis: Scaffold extraction, chemotype profiling, structural diversity - Clustering: Molecular clustering using various methods (k-means, hierarchical, DBSCAN) - SAR Analysis: Structure-Activity Relationship analysis, activity cliffs, matched molecular pairs - Similarity Analysis: Molecular similarity, diversity metrics, nearest neighbor searches

GTM-Integrated Design: - Primary use case: Downstream analysis after GTM agents (nodes as clusters) - Also works with ANY data source: t-SNE clusters, user CSVs, ChEMBL families - Standardized input: DataFrame with 'smiles' + optional 'cluster_id' + optional 'activity' - Produces structured data output (DataFrames, dicts) - NO report generation - Report generation handled by separate ReportGeneratorAgent

Tools: - ChemicalSimilarityToolkit: Fingerprints, similarity metrics, scaffold extraction - PointerPandasTools: DataFrame operations with S3 support - GTMToolkit: Access to GTM data (source_mols, node projections)

Source code in src/cs_copilot/agents/factories.py 
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
class ChemoinformaticianFactory(BaseAgentFactory):
    """Factory for creating comprehensive chemoinformatics analysis agents.

    This agent is a versatile chemoinformatician capable of:
    - **Chemotype Analysis**: Scaffold extraction, chemotype profiling, structural diversity
    - **Clustering**: Molecular clustering using various methods (k-means, hierarchical, DBSCAN)
    - **SAR Analysis**: Structure-Activity Relationship analysis, activity cliffs, matched molecular pairs
    - **Similarity Analysis**: Molecular similarity, diversity metrics, nearest neighbor searches

    GTM-Integrated Design:
    - Primary use case: Downstream analysis after GTM agents (nodes as clusters)
    - Also works with ANY data source: t-SNE clusters, user CSVs, ChEMBL families
    - Standardized input: DataFrame with 'smiles' + optional 'cluster_id' + optional 'activity'
    - Produces structured data output (DataFrames, dicts) - NO report generation
    - Report generation handled by separate ReportGeneratorAgent

    Tools:
    - ChemicalSimilarityToolkit: Fingerprints, similarity metrics, scaffold extraction
    - PointerPandasTools: DataFrame operations with S3 support
    - GTMToolkit: Access to GTM data (source_mols, node projections)
    """

    agent_type = "chemoinformatician"

    def get_agent_config(self) -> AgentConfig:
        return AgentConfig(
            name="chemoinformatician_agent",
            description=CHEMOINFORMATICIAN_DESCRIPTION,
            tools=[
                ChemicalSimilarityToolkit(),
                PointerPandasTools(),
                SessionMemoryToolkit(),
                GTMToolkit(),  # Enable GTM data access for downstream analysis
                SkillToolkit(),
                # Future: QSARToolkit, ClusteringToolkit, DescriptorToolkit
            ],
            instructions=CHEMOINFORMATICIAN_INSTRUCTIONS,
            session_state={
                # Normalized input data for analysis
                "analysis_input": None,  # DataFrame with standardized columns (smiles, cluster_id?, activity?)
                # Chemotype/Scaffold Analysis
                "chemotype_analysis": {
                    "scaffolds_per_cluster": None,
                    "similarity_matrix": None,
                    "summary_stats": None,
                    "metadata": {},
                    "output_paths": {
                        "scaffolds_csv": None,
                        "similarity_csv": None,
                    },
                },
                # Clustering Analysis
                "clustering_results": {
                    "cluster_assignments": None,  # DataFrame with cluster_id column
                    "cluster_metrics": None,  # Silhouette, Davies-Bouldin, etc.
                    "cluster_centroids": None,
                    "method": None,  # 'gtm', 'kmeans', 'dbscan', 'hierarchical', etc.
                },
                # SAR Analysis
                "sar_analysis": {
                    "activity_cliffs": None,  # Detected activity cliffs
                    "mmps": None,  # Matched molecular pairs
                    "series_analysis": None,  # Chemical series breakdown
                    "potency_trends": None,
                },
                # Similarity/Diversity
                "similarity_analysis": {
                    "similarity_matrix": None,
                    "diversity_metrics": None,
                    "nearest_neighbors": None,
                },
                # General data paths
                "analysis_outputs": {
                    "primary_data_csv": None,
                    "supplementary_data": [],
                },
            },
        )

MolecularDesignerFactory

Bases: BaseAgentFactory

Factory for creating small-molecule design agents.

Supports two modes: - Engine-driven design: Use autoencoder or LLM engines behind a common facade - Standalone autoencoder: Encode/decode SMILES, sample latent space, interpolate, explore neighborhoods - GTM-guided: Combine GTM maps with generative engines for targeted molecular design from specific map regions (by density, activity, or coordinates)

Enhanced with GTM cache awareness to avoid redundant GTM loading when working with GTM Agent in the same session.

Source code in src/cs_copilot/agents/factories.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
class MolecularDesignerFactory(BaseAgentFactory):
    """Factory for creating small-molecule design agents.

    Supports two modes:
    - **Engine-driven design**: Use autoencoder or LLM engines behind a common facade
    - **Standalone autoencoder**: Encode/decode SMILES, sample latent space, interpolate, explore neighborhoods
    - **GTM-guided**: Combine GTM maps with generative engines for targeted molecular design
      from specific map regions (by density, activity, or coordinates)

    Enhanced with GTM cache awareness to avoid redundant GTM loading when working with GTM Agent
    in the same session.
    """

    agent_type = "molecular_designer"

    def get_agent_config(self) -> AgentConfig:
        autoencoder_toolkit: AutoencoderToolkit = make_autoencoder_toolkit()
        return AgentConfig(
            name="molecular_designer_agent",
            description=MOLECULAR_DESIGNER_DESCRIPTION,
            tools=[
                MolecularDesignerToolkit(autoencoder_toolkit=autoencoder_toolkit),
                autoencoder_toolkit,
                GTMToolkit(),
                ChemicalSimilarityToolkit(),
                PointerPandasTools(),
                SessionMemoryToolkit(),
                SkillToolkit(),
            ],
            instructions=MOLECULAR_DESIGNER_INSTRUCTIONS,
            session_state={
                "data_file_paths": {
                    "dataset_path": None,  # Backward-compatible alias for clean_dataset_path.
                    "raw_dataset_path": None,
                    "clean_dataset_path": None,
                    "filtered_dataset_path": None,
                    "descriptor_parquet_path": None,
                    "standardization_report_path": None,
                },
            },
        )

GTMAgentFactory

Bases: BaseAgentFactory

Factory for creating unified GTM agents (consolidates optimization, loading, density, activity, projection).

This factory creates a single agent that handles all GTM-related operations via mode-based dispatch: - optimize: Build and optimize new GTM maps - load: Load existing GTM models from S3/local/HuggingFace - density: Analyze compound distributions and neighborhood preservation - activity: Create activity landscapes for SAR analysis - project: Project external datasets onto existing GTM maps

Features smart caching to avoid redundant GTM loading across operations.

Source code in src/cs_copilot/agents/factories.py 
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
class GTMAgentFactory(BaseAgentFactory):
    """Factory for creating unified GTM agents (consolidates optimization, loading, density, activity, projection).

    This factory creates a single agent that handles all GTM-related operations via mode-based dispatch:
    - optimize: Build and optimize new GTM maps
    - load: Load existing GTM models from S3/local/HuggingFace
    - density: Analyze compound distributions and neighborhood preservation
    - activity: Create activity landscapes for SAR analysis
    - project: Project external datasets onto existing GTM maps

    Features smart caching to avoid redundant GTM loading across operations.
    """

    agent_type = "gtm_agent"

    def get_agent_config(self) -> AgentConfig:
        return AgentConfig(
            name="gtm_agent",
            description=GTM_AGENT_DESCRIPTION,
            tools=[
                GTMToolkit(),
                PointerPandasTools(),
                SessionMemoryToolkit(),
                save_gtm_landscape_plot,
                save_gtm_plot,
                SkillToolkit(),
            ],
            instructions=GTM_AGENT_INSTRUCTIONS,
            session_state={
                "gtm_cache": {
                    "model": None,
                    "dataset": None,
                    "metadata": {
                        "optimization_strategy": None,
                    },
                },
                "gtm_file_paths": {
                    "gtm_path": None,
                    "dataset_path": None,
                    "gtm_plot_path": None,
                },
                "analysis_results": {
                    "density_csv": None,
                    "activity_csv": None,
                    "projection_csv": None,
                    "plots": [],
                },
                "landscape_files": {  # Backward compatibility
                    "landscape_data_csv": None,
                    "landscape_plot": None,
                },
            },
        )

ReportGeneratorFactory

Bases: BaseAgentFactory

Factory for creating report generation agents.

This agent handles ALL report generation and visualization across different analysis types: - Chemotype analysis reports - GTM density reports - GTM activity/SAR reports - Molecular designer generation reports - Combined/custom reports

Separation of Concerns: Analysis agents produce structured data, Report Generator handles presentation.

This architecture enables: - Consistent formatting across all report types - Reusable visualization patterns - Easy updates to report styles (change in one place) - Clean separation: data processing vs visualization/formatting

Source code in src/cs_copilot/agents/factories.py 
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
class ReportGeneratorFactory(BaseAgentFactory):
    """Factory for creating report generation agents.

    This agent handles ALL report generation and visualization across different analysis types:
    - Chemotype analysis reports
    - GTM density reports
    - GTM activity/SAR reports
    - Molecular designer generation reports
    - Combined/custom reports

    **Separation of Concerns**: Analysis agents produce structured data, Report Generator handles presentation.

    This architecture enables:
    - Consistent formatting across all report types
    - Reusable visualization patterns
    - Easy updates to report styles (change in one place)
    - Clean separation: data processing vs visualization/formatting
    """

    agent_type = "report_generator"

    def get_agent_config(self) -> AgentConfig:
        return AgentConfig(
            name="report_generator_agent",
            description=REPORT_GENERATOR_DESCRIPTION,
            tools=[
                PointerPandasTools(),
                SessionMemoryToolkit(),
                save_gtm_landscape_plot,  # For saved GTM landscape tables
                save_gtm_plot,  # For GTM-specific visualizations
                get_report_evidence,
                save_rich_report,  # Persists image-rich HTML/PDF reports
                save_markdown_report,  # Persists the final markdown report
                SkillToolkit(),
                # Plotting libraries (matplotlib, seaborn) available via Python environment
            ],
            instructions=REPORT_GENERATOR_INSTRUCTIONS,
            session_state={
                "report_outputs": {
                    "report_path": None,
                    "report_paths": {},
                    "plots": [],
                    "report_type": None,
                },
            },
        )

RobustnessEvaluationFactory

Bases: BaseAgentFactory

Factory for creating robustness test evaluation agents.

Source code in src/cs_copilot/agents/factories.py 
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
class RobustnessEvaluationFactory(BaseAgentFactory):
    """Factory for creating robustness test evaluation agents."""

    agent_type = "robustness_evaluation"

    def get_agent_config(self) -> AgentConfig:
        return AgentConfig(
            name="robustness_evaluator_agent",
            description=ROBUSTNESS_EVALUATION_DESCRIPTION,
            tools=[
                PointerPandasTools(),
                SessionMemoryToolkit(),
                RobustnessAnalysisToolkit(),
                SkillToolkit(),
            ],
            instructions=ROBUSTNESS_EVALUATION_INSTRUCTIONS,
            session_state={
                "loaded_results": {},
                "analysis_outputs": {
                    "summary_report": None,
                    "comparison_report": None,
                    "recommendations": None,
                },
            },
        )

SynPlannerFactory

Bases: BaseAgentFactory

Factory for creating retrosynthetic planning agents powered by SynPlanner.

This agent wraps the official SynPlanner package to perform retrosynthetic analysis on target molecules. It accepts SMILES strings or molecule names, resolves them to canonical SMILES (via PubChem / RDKit), runs the MCTS-based retrosynthesis search, and returns structured route descriptions with optional SVG/PNG visualizations.

Source code in src/cs_copilot/agents/factories.py 
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
class SynPlannerFactory(BaseAgentFactory):
    """Factory for creating retrosynthetic planning agents powered by SynPlanner.

    This agent wraps the official SynPlanner package to perform retrosynthetic
    analysis on target molecules.  It accepts SMILES strings or molecule names,
    resolves them to canonical SMILES (via PubChem / RDKit), runs the MCTS-based
    retrosynthesis search, and returns structured route descriptions with
    optional SVG/PNG visualizations.
    """

    agent_type = "synplanner"

    def get_agent_config(self) -> AgentConfig:
        return AgentConfig(
            name="synplanner_agent",
            description=SYNPLANNER_DESCRIPTION,
            tools=[
                SynPlannerToolkit(),
                SkillToolkit(),
            ],
            instructions=SYNPLANNER_INSTRUCTIONS,
        )

PeptideDesignerFactory

Bases: BaseAgentFactory

Factory for creating peptide design agents.

This agent exposes a Peptide Designer facade over multiple peptide design engines. The default WAE engine encodes, decodes, samples, and interpolates amino acid sequences; the LLM engine proposes sequence candidates from natural-language objectives. The WAE model can generate any peptides; activity landscape data comes from DBAASP (antimicrobial peptides specifically).

Key capabilities: - Encoding: Convert peptide sequences to 100-dimensional latent vectors - Decoding: Generate peptide sequences from latent vectors - Sampling: Generate novel peptides from Gaussian prior - Interpolation: Smooth transitions between peptides in latent space - Neighborhood exploration: Generate peptide analogs - GTM integration: Train GTMs on latent space, create activity landscapes - Activity landscapes: Use DBAASP data (specific to antimicrobial peptides)

Input format: Space-separated single-letter amino acid codes Example: "M L L L L L A L A L L A L L L A L L L"

Source code in src/cs_copilot/agents/factories.py 
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
class PeptideDesignerFactory(BaseAgentFactory):
    """Factory for creating peptide design agents.

    This agent exposes a Peptide Designer facade over multiple peptide design
    engines. The default WAE engine encodes, decodes, samples, and interpolates
    amino acid sequences; the LLM engine proposes sequence candidates from
    natural-language objectives. The WAE model can generate any peptides;
    activity landscape data comes from DBAASP (antimicrobial peptides specifically).

    Key capabilities:
    - **Encoding**: Convert peptide sequences to 100-dimensional latent vectors
    - **Decoding**: Generate peptide sequences from latent vectors
    - **Sampling**: Generate novel peptides from Gaussian prior
    - **Interpolation**: Smooth transitions between peptides in latent space
    - **Neighborhood exploration**: Generate peptide analogs
    - **GTM integration**: Train GTMs on latent space, create activity landscapes
    - **Activity landscapes**: Use DBAASP data (specific to antimicrobial peptides)

    Input format: Space-separated single-letter amino acid codes
    Example: "M L L L L L A L A L L A L L L A L L L"
    """

    agent_type = "peptide_designer"

    def get_agent_config(self) -> AgentConfig:
        return AgentConfig(
            name="peptide_designer_agent",
            description=PEPTIDE_DESIGNER_DESCRIPTION,
            tools=[
                PeptideDesignerToolkit(),
                GTMToolkit(),
                PointerPandasTools(),
                SessionMemoryToolkit(),
                save_gtm_landscape_plot,
                save_gtm_plot,
                SkillToolkit(),
            ],
            instructions=PEPTIDE_DESIGNER_INSTRUCTIONS,
        )

SingleAgentFactory

Bases: BaseAgentFactory

Factory for the single-agent baseline used in the multi-agent ablation.

One flat Agno Agent that holds the UNION of the seven team specialists' toolkits (ChEMBL, GTM, chemoinformatics, molecular + peptide design, retrosynthesis, reporting) with no coordinator, no routing, and no per-specialist context isolation. It is deliberately NOT added to the runtime team (teams.py stays 7 members); it is constructed separately for the robustness comparison so the only variable vs the team is the agentic structure (same model, same tools, same tasks).

Agno registers toolkit methods by name and keeps the first-registered on a clash. The peptide toolkit is therefore registered under its MCP-style peptide_* namespace in this flat context. Specialist agents retain their native unprefixed names because their tool contexts are already isolated.

Source code in src/cs_copilot/agents/factories.py 
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
class SingleAgentFactory(BaseAgentFactory):
    """Factory for the single-agent baseline used in the multi-agent ablation.

    One flat Agno ``Agent`` that holds the UNION of the seven team specialists'
    toolkits (ChEMBL, GTM, chemoinformatics, molecular + peptide design,
    retrosynthesis, reporting) with no coordinator, no routing, and no
    per-specialist context isolation. It is deliberately NOT added to the runtime
    team (``teams.py`` stays 7 members); it is constructed separately for the
    robustness comparison so the only variable vs the team is the agentic
    structure (same model, same tools, same tasks).

    Agno registers toolkit methods by name and keeps the first-registered on a
    clash. The peptide toolkit is therefore registered under its MCP-style
    ``peptide_*`` namespace in this flat context. Specialist agents retain their
    native unprefixed names because their tool contexts are already isolated.
    """

    agent_type = "single_agent"

    def get_agent_config(self) -> AgentConfig:
        autoencoder_toolkit: AutoencoderToolkit = make_autoencoder_toolkit()
        return AgentConfig(
            name="single_agent",
            description=SINGLE_AGENT_DESCRIPTION,
            tools=[
                # Data + space
                ChemblToolkit(),
                GTMToolkit(),
                ChemicalSimilarityToolkit(),
                # Design engines. The peptide namespace prevents its chemically
                # distinct methods from colliding with the molecular toolkits.
                MolecularDesignerToolkit(autoencoder_toolkit=autoencoder_toolkit),
                autoencoder_toolkit,
                PeptideDesignerToolkit(tool_name_prefix="peptide_"),
                SynPlannerToolkit(),
                # Shared infrastructure (one instance each).
                SessionMemoryToolkit(),
                PointerPandasTools(),
                SkillToolkit(),
                # Plot / report callables.
                save_gtm_landscape_plot,
                save_gtm_plot,
                get_report_evidence,
                save_rich_report,
                save_markdown_report,
            ],
            instructions=SINGLE_AGENT_INSTRUCTIONS,
            # Union of the team members' session_state defaults so the flat agent
            # carries the same artifact/analysis slots the team accumulates. Keys
            # mirror ChEMBL/MolecularDesigner (data_file_paths), Chemoinformatician
            # (analysis_input, chemotype_analysis, clustering_results, sar_analysis,
            # similarity_analysis, analysis_outputs), GTM (gtm_cache, gtm_file_paths,
            # analysis_results, landscape_files), and Report (report_outputs).
            # A drift guard in tests/unit/test_single_agent_factory.py checks these
            # stay in sync with the member factories.
            session_state={
                "data_file_paths": {
                    "dataset_path": None,  # Backward-compatible alias for clean_dataset_path.
                    "raw_dataset_path": None,
                    "clean_dataset_path": None,
                    "filtered_dataset_path": None,
                    "descriptor_parquet_path": None,
                    "standardization_report_path": None,
                },
                "analysis_input": None,
                "chemotype_analysis": {
                    "scaffolds_per_cluster": None,
                    "similarity_matrix": None,
                    "summary_stats": None,
                    "metadata": {},
                    "output_paths": {
                        "scaffolds_csv": None,
                        "similarity_csv": None,
                    },
                },
                "clustering_results": {
                    "cluster_assignments": None,
                    "cluster_metrics": None,
                    "cluster_centroids": None,
                    "method": None,
                },
                "sar_analysis": {
                    "activity_cliffs": None,
                    "mmps": None,
                    "series_analysis": None,
                    "potency_trends": None,
                },
                "similarity_analysis": {
                    "similarity_matrix": None,
                    "diversity_metrics": None,
                    "nearest_neighbors": None,
                },
                "gtm_cache": {
                    "model": None,
                    "dataset": None,
                    "metadata": {
                        "optimization_strategy": None,
                    },
                },
                "gtm_file_paths": {
                    "gtm_path": None,
                    "dataset_path": None,
                    "gtm_plot_path": None,
                },
                "analysis_results": {
                    "density_csv": None,
                    "activity_csv": None,
                    "projection_csv": None,
                    "plots": [],
                },
                "landscape_files": {
                    "landscape_data_csv": None,
                    "landscape_plot": None,
                },
                "report_outputs": {
                    "report_path": None,
                    "report_paths": {},
                    "plots": [],
                    "report_type": None,
                },
                "analysis_outputs": {
                    "primary_data_csv": None,
                    "supplementary_data": [],
                },
            },
        )

instructions

Agent behavioral instructions for cs_copilot.

These lists answer how each agent behaves — the rules, policies, and conventions that Agno injects into the system message via Agent(instructions=...). They are deliberately kept separate from the persona identities in descriptions.py (Agno's description=) and from the mutable multi-tool procedures in the skill/workflow catalogs. Each agent's instructions point at the catalog and expect the agent to fetch and follow the procedure there rather than embedding step-by-step tool sequences here.

registry

Agent registry system for managing and creating agents dynamically. Provides the main public API for agent creation.

AgentRegistry

Registry for managing agent factories and configurations.

Source code in src/cs_copilot/agents/registry.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class AgentRegistry:
    """Registry for managing agent factories and configurations."""

    def __init__(self):
        self._factories: Dict[str, BaseAgentFactory] = {}
        self._aliases: Dict[str, str] = {}  # Alias -> canonical agent_type mapping
        self.logger = logging.getLogger(__name__)

    def register(
        self, agent_type: str, factory: BaseAgentFactory, aliases: List[str] = None
    ) -> None:
        """Register an agent factory with optional aliases.

        Args:
            agent_type: Canonical agent type name
            factory: Factory instance
            aliases: Optional list of alias names that redirect to this agent
        """
        if agent_type in self._factories:
            self.logger.warning(f"Overriding existing factory for agent type: {agent_type}")
        self._factories[agent_type] = factory
        self.logger.info(f"Registered factory for agent type: {agent_type}")

        # Register aliases
        if aliases:
            for alias in aliases:
                self._aliases[alias] = agent_type
                self.logger.info(f"Registered alias '{alias}' -> '{agent_type}'")

    def create_agent(self, agent_type: str, model: Model, **kwargs: Any) -> Agent:
        """Create an agent by type (supports aliases).

        Args:
            agent_type: Agent type or alias
            model: LLM model instance
            **kwargs: Additional arguments for agent creation

        Returns:
            Agent instance

        Raises:
            ValueError: If agent_type/alias is not registered
        """
        # Resolve alias if provided
        resolved_type = self._aliases.get(agent_type, agent_type)

        if resolved_type not in self._factories:
            available_types = list(self._factories.keys())
            available_aliases = list(self._aliases.keys())
            raise ValueError(
                f"Unknown agent type: {agent_type}. "
                f"Available types: {available_types}. "
                f"Available aliases: {available_aliases}"
            )

        factory = self._factories[resolved_type]
        return factory.create_agent(model, **kwargs)

    def list_agent_types(self) -> List[str]:
        """List all registered agent types."""
        return list(self._factories.keys())

    def auto_register(self) -> None:
        """Automatically discover and register all available factories."""
        for _, cls in inspect.getmembers(factory_module, inspect.isclass):
            if (
                issubclass(cls, BaseAgentFactory)
                and cls is not BaseAgentFactory
                and getattr(cls, "agent_type", None)
            ):
                # Get optional aliases from factory class
                aliases = getattr(cls, "aliases", None)
                self.register(cls.agent_type, cls(), aliases=aliases)
register(agent_type, factory, aliases=None)

Register an agent factory with optional aliases.

Parameters:

Name Type Description Default
agent_type str

Canonical agent type name

 required
factory BaseAgentFactory

Factory instance

 required
aliases List[str]

Optional list of alias names that redirect to this agent

 None
Source code in src/cs_copilot/agents/registry.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def register(
    self, agent_type: str, factory: BaseAgentFactory, aliases: List[str] = None
) -> None:
    """Register an agent factory with optional aliases.

    Args:
        agent_type: Canonical agent type name
        factory: Factory instance
        aliases: Optional list of alias names that redirect to this agent
    """
    if agent_type in self._factories:
        self.logger.warning(f"Overriding existing factory for agent type: {agent_type}")
    self._factories[agent_type] = factory
    self.logger.info(f"Registered factory for agent type: {agent_type}")

    # Register aliases
    if aliases:
        for alias in aliases:
            self._aliases[alias] = agent_type
            self.logger.info(f"Registered alias '{alias}' -> '{agent_type}'")
create_agent(agent_type, model, **kwargs)

Create an agent by type (supports aliases).

Parameters:

Name Type Description Default
agent_type str

Agent type or alias

 required
model Model

LLM model instance

 required
**kwargs Any

Additional arguments for agent creation

 {}

Returns:

Type Description
Agent

Agent instance

Raises:

Type Description
ValueError

If agent_type/alias is not registered
Source code in src/cs_copilot/agents/registry.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def create_agent(self, agent_type: str, model: Model, **kwargs: Any) -> Agent:
    """Create an agent by type (supports aliases).

    Args:
        agent_type: Agent type or alias
        model: LLM model instance
        **kwargs: Additional arguments for agent creation

    Returns:
        Agent instance

    Raises:
        ValueError: If agent_type/alias is not registered
    """
    # Resolve alias if provided
    resolved_type = self._aliases.get(agent_type, agent_type)

    if resolved_type not in self._factories:
        available_types = list(self._factories.keys())
        available_aliases = list(self._aliases.keys())
        raise ValueError(
            f"Unknown agent type: {agent_type}. "
            f"Available types: {available_types}. "
            f"Available aliases: {available_aliases}"
        )

    factory = self._factories[resolved_type]
    return factory.create_agent(model, **kwargs)
list_agent_types()

List all registered agent types.

Source code in src/cs_copilot/agents/registry.py 
77
78
79
def list_agent_types(self) -> List[str]:
    """List all registered agent types."""
    return list(self._factories.keys())
auto_register()

Automatically discover and register all available factories.

Source code in src/cs_copilot/agents/registry.py 
81
82
83
84
85
86
87
88
89
90
91
def auto_register(self) -> None:
    """Automatically discover and register all available factories."""
    for _, cls in inspect.getmembers(factory_module, inspect.isclass):
        if (
            issubclass(cls, BaseAgentFactory)
            and cls is not BaseAgentFactory
            and getattr(cls, "agent_type", None)
        ):
            # Get optional aliases from factory class
            aliases = getattr(cls, "aliases", None)
            self.register(cls.agent_type, cls(), aliases=aliases)

create_agent(agent_type, model, **kwargs)

Create an agent by type using the global registry.

Parameters:

Name Type Description Default
agent_type str

The type of agent to create

 required
model Model

The language model to use

 required
**kwargs Any

Additional arguments passed to the agent factory

 {}

Returns:

Name Type Description
Agent Agent

The created agent instance

Raises:

Type Description
ValueError

If agent_type is not registered
AgentCreationError

If agent creation fails
Source code in src/cs_copilot/agents/registry.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def create_agent(agent_type: str, model: Model, **kwargs: Any) -> Agent:
    """
    Create an agent by type using the global registry.

    Args:
        agent_type: The type of agent to create
        model: The language model to use
        **kwargs: Additional arguments passed to the agent factory

    Returns:
        Agent: The created agent instance

    Raises:
        ValueError: If agent_type is not registered
        AgentCreationError: If agent creation fails
    """
    return _agent_registry.create_agent(agent_type, model, **kwargs)

list_available_agent_types()

List all available agent types.

Source code in src/cs_copilot/agents/registry.py 
118
119
120
def list_available_agent_types() -> List[str]:
    """List all available agent types."""
    return _agent_registry.list_agent_types()

get_registry()

Get the global agent registry instance.

Source code in src/cs_copilot/agents/registry.py 
123
124
125
def get_registry() -> AgentRegistry:
    """Get the global agent registry instance."""
    return _agent_registry

session_runs

One durable v2 workflow run per in-process chat.

Chainlit and the CLI keep conversation history in Agno's SQLite store; the scientific state of a chat — tool events, registered artifacts, handoffs — is recorded in one ad-hoc agno-session workflow run per chat. The run is created lazily before the first message is processed, reused when a chat resumes, and never completed automatically because a chat can always resume.

ensure_agno_session_run(team, *, session_id, mode=None)

Return the chat's durable run, creating and binding it on first use.

Returns None when the execution mode is off. Idempotent: once a run is attached to team it is returned as is.

Source code in src/cs_copilot/agents/session_runs.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def ensure_agno_session_run(
    team: Any,
    *,
    session_id: str,
    mode: ExecutionMode | str | None = None,
) -> RunContext | None:
    """Return the chat's durable run, creating and binding it on first use.

    Returns ``None`` when the execution mode is ``off``. Idempotent: once a
    run is attached to ``team`` it is returned as is.
    """

    selected = execution_mode_from_env(mode if mode is not None else _team_mode(team))
    if selected is ExecutionMode.OFF:
        return None
    binding = get_binding(team)
    if binding is not None and binding.run_context is not None:
        return binding.run_context

    state = _team_state(team)
    enforce = selected is ExecutionMode.ENFORCE
    run_context = _load_chat_run(state, session_id, verify_artifacts=enforce)
    if run_context is None:
        previous = state.pop(OUTPUT_CONTEXT_KEY, None)
        if isinstance(previous, Mapping):
            legacy = state.setdefault("legacy_output_contexts", [])
            if isinstance(legacy, list):
                legacy.append(dict(previous))
        run_context = RunContext.create(
            AGNO_SESSION_SLUG,
            session_state=state,
            session_id=session_id,
        )
        run_context.transition_run(RunStatus.RUNNING, reason="chat session started")
    else:
        run_context.bind_session_state(state)
        reconcile_interrupted_work(run_context)
    if enforce:
        adopt_session_uploads(run_context, state)

    _publish_run_identity(team, state, run_context)
    attach_execution(team, run_context=run_context, mode=selected)
    return run_context

restore_team_session_state(team, saved_state)

Restore a persisted chat state into the team's shared dict in place.

Members hold a reference to the team's session_state dict; replacing the dict would leave them writing to a stale copy. Saved values win, except for the team-owned runtime keys of the freshly built team.

Source code in src/cs_copilot/agents/session_runs.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def restore_team_session_state(team: Any, saved_state: Mapping[str, Any]) -> dict[str, Any]:
    """Restore a persisted chat state into the team's shared dict in place.

    Members hold a reference to the team's ``session_state`` dict; replacing
    the dict would leave them writing to a stale copy. Saved values win,
    except for the team-owned runtime keys of the freshly built team.
    """

    state = _team_state(team)
    preserved = {key: state[key] for key in _TEAM_OWNED_KEYS if key in state}
    state.clear()
    state.update(dict(saved_state))
    state.update(preserved)
    return state

finalize_agno_turn(team, *, failed=False)

Close the ad-hoc tasks delegated during the turn that just ended.

Source code in src/cs_copilot/agents/session_runs.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def finalize_agno_turn(team: Any, *, failed: bool = False) -> None:
    """Close the ad-hoc tasks delegated during the turn that just ended."""

    binding = get_binding(team)
    if binding is None or binding.run_context is None:
        return
    run_context = binding.run_context
    for task_id in binding.take_turn_tasks():
        try:
            task = run_context.refresh().tasks.get(task_id)
            if task is None or task.status is not TaskStatus.RUNNING:
                continue
            if run_context.pending_tool_invocations(task_id=task_id, domain_only=True):
                continue
            if failed:
                run_context.transition_task(
                    task_id,
                    TaskStatus.FAILED,
                    error=ToolError(
                        code="internal",
                        message="the chat turn failed before the task finished",
                        retryable=True,
                    ),
                )
            else:
                run_context.transition_task(task_id, TaskStatus.COMPLETED)
        except Exception as exc:  # noqa: BLE001 - bookkeeping must not fail a turn
            logger.warning("Could not finalize ad-hoc task %s: %s", task_id, exc)
            binding.note_problem(f"finalize {task_id}: {exc}")

reconcile_interrupted_work(run_context)

Close tool spans and tasks left open by a previous process of this chat.

A span recorded by this runtime whose owning process is not this one can no longer finish: the chat server restarted or crashed. It is marked abandoned so it no longer blocks run, task, and artifact transitions, and tasks still marked running are failed as interrupted.

Source code in src/cs_copilot/agents/session_runs.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
def reconcile_interrupted_work(run_context: RunContext) -> list[str]:
    """Close tool spans and tasks left open by a previous process of this chat.

    A span recorded by this runtime whose owning process is not this one can
    no longer finish: the chat server restarted or crashed. It is marked
    abandoned so it no longer blocks run, task, and artifact transitions, and
    tasks still marked running are failed as interrupted.
    """

    abandoned: list[str] = []
    for span_id, owner in _open_agno_spans(run_context):
        if owner == PROCESS_OWNER:
            continue
        try:
            run_context.abandon_tool_invocation(
                span_id,
                reason="the chat process that ran this tool call ended before it finished",
            )
            abandoned.append(span_id)
        except Exception as exc:  # noqa: BLE001
            logger.warning("Could not abandon orphaned tool span %s: %s", span_id, exc)
    run = run_context.refresh()
    for task in list(run.tasks.values()):
        if task.status is not TaskStatus.RUNNING:
            continue
        if run_context.pending_tool_invocations(task_id=task.task_id, domain_only=True):
            continue
        try:
            run_context.transition_task(
                task.task_id,
                TaskStatus.FAILED,
                error=ToolError(
                    code="internal",
                    message="the chat session was interrupted while this task was running",
                    retryable=True,
                ),
            )
        except Exception as exc:  # noqa: BLE001
            logger.warning("Could not fail interrupted task %s: %s", task.task_id, exc)
    return abandoned

store_chat_upload(run_context, *, filename, content)

Write a user upload inside the chat's run and register it as untrusted.

Returns (storage_path, artifact_id). Identical re-uploads reuse the same content-addressed file and artifact.

Source code in src/cs_copilot/agents/session_runs.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def store_chat_upload(
    run_context: RunContext,
    *,
    filename: str,
    content: bytes,
) -> tuple[str, str]:
    """Write a user upload inside the chat's run and register it as untrusted.

    Returns ``(storage_path, artifact_id)``. Identical re-uploads reuse the
    same content-addressed file and artifact.
    """

    name = PurePosixPath(str(filename).replace("\\", "/")).name or "upload"
    stem, dot, extension = name.partition(".")
    digest = hashlib.sha256(content).hexdigest()
    safe_stem = "".join(ch if ch.isalnum() or ch in "-_" else "_" for ch in stem)[:80] or "upload"
    relative = f"inputs/uploads/{safe_stem}-{digest[:8]}{dot}{extension}"
    storage_key = run_context.layout.artifact_rel_path(relative)
    existing = next(
        (
            record
            for record in run_context.run.artifacts.values()
            if record.relative_path == relative
        ),
        None,
    )
    if existing is None:
        with S3.open(storage_key, "wb") as handle:
            handle.write(content)
    record = run_context.register_artifact(
        relative,
        artifact_type="user_upload",
        mime_type=mimetypes.guess_type(name)[0] or "application/octet-stream",
        producer_tool="chat_upload",
        provenance={"registration": "manual", "source": "chat_upload", "original_name": name},
        trust="untrusted",
    )
    return S3.path(storage_key), record.artifact_id

adopt_session_uploads(run_context, state)

Move uploads stored outside the run (older chats) into it as artifacts.

In enforce mode tools may only read registered artifacts, so uploads that predate the chat's run are copied into it and registered; the original files are left in place.

Source code in src/cs_copilot/agents/session_runs.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def adopt_session_uploads(run_context: RunContext, state: dict[str, Any]) -> list[str]:
    """Move uploads stored outside the run (older chats) into it as artifacts.

    In enforce mode tools may only read registered artifacts, so uploads that
    predate the chat's run are copied into it and registered; the original
    files are left in place.
    """

    uploaded = state.get("uploaded_files")
    if not isinstance(uploaded, dict):
        return []
    artifacts = state.setdefault("uploaded_artifacts", {})
    run_prefix = S3.path(run_context.layout.run_root).rstrip("/") + "/"
    adopted: list[str] = []
    for name, path in list(uploaded.items()):
        if not isinstance(path, str) or path.startswith(run_prefix) or name in artifacts:
            continue
        try:
            with S3.open(path, "rb") as handle:
                content = handle.read()
            stored, artifact_id = store_chat_upload(run_context, filename=name, content=content)
        except Exception as exc:  # noqa: BLE001
            logger.warning("Could not adopt upload %s into the chat run: %s", name, exc)
            continue
        uploaded[name] = stored
        artifacts[name] = artifact_id
        adopted.append(name)
    return adopted

single_agent

Single-agent baseline builder for the multi-agent ablation.

Mirrors :func:cs_copilot.agents.teams.get_cs_copilot_agent_team but returns ONE flat Agno Agent holding the union of all specialist toolkits. Use the same Model instance for both this builder and get_cs_copilot_agent_team so the only variable in the comparison is the agentic structure (single vs multi), not the base model, tools, or harness.

get_cs_copilot_single_agent(model, *, markdown=True, debug_mode=False, enable_mlflow_tracking=True)

Create the single flat agent (all toolkits, no team, no routing).

Parameters:

Name Type Description Default
model Model

Agno Model instance — pass the SAME instance used for the team so the ablation holds the model constant.

 required
markdown bool

Format output in markdown (matches team members).

 True
debug_mode bool

Enable debug logs.

 False
enable_mlflow_tracking bool

Wrap run/arun with MLflow tracking, exactly as team members are wrapped, for a fair comparison.

 True

Returns:

Name Type Description
Agent Agent

The configured single-agent baseline. Exposes .run /
Agent

.get_session_state like the team, so the robustness harness can drive
Agent

either arm through one interface.
Source code in src/cs_copilot/agents/single_agent.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def get_cs_copilot_single_agent(
    model: Model,
    *,
    markdown: bool = True,
    debug_mode: bool = False,
    enable_mlflow_tracking: bool = True,
) -> Agent:
    """Create the single flat agent (all toolkits, no team, no routing).

    Args:
        model: Agno Model instance — pass the SAME instance used for the team so
            the ablation holds the model constant.
        markdown: Format output in markdown (matches team members).
        debug_mode: Enable debug logs.
        enable_mlflow_tracking: Wrap ``run``/``arun`` with MLflow tracking, exactly
            as team members are wrapped, for a fair comparison.

    Returns:
        Agent: The configured single-agent baseline. Exposes ``.run`` /
        ``.get_session_state`` like the team, so the robustness harness can drive
        either arm through one interface.
    """
    logger = logging.getLogger(__name__)
    logger.info("Creating Cs_copilot single-agent baseline")

    # Seed the same shared-state contract the team members get (teams.py:70-84):
    # resource profile + a scratch dict. The factory's union session_state is then
    # deep-merged in by BaseAgentFactory.create_agent without overwriting these.
    seed_session_state = {
        "resource_profile": analyze_resources(),
        "agent_scratch": {},
    }

    agent = create_agent(
        "single_agent",
        model=model,
        markdown=markdown,
        debug_mode=debug_mode,
        enable_mlflow_tracking=enable_mlflow_tracking,
        session_state=seed_session_state,
    )
    logger.info("Successfully created Cs_copilot single-agent baseline")
    return agent

teams

Team coordination functionality for multi-agent workflows.

get_cs_copilot_agent_team(model, *, markdown=True, debug_mode=False, show_members_responses=True, enable_memory=True, db_file=None, enable_mlflow_tracking=True, run_context=None, execution_mode=None)

Create a coordinated team of cs_copilot agents using Agno.

Parameters:

Name Type Description Default
model Model

Agno Model instance used for team coordination and member agents

 required
markdown bool

Format output in markdown

 True
debug_mode bool

Enable debug logs

 False
show_members_responses bool

Print member responses during coordination

 True
enable_memory bool

Enable persistent session history (default: True). Cross-session user/agentic memories stay disabled to prevent state leakage.

 True
db_file str

Custom database file path. If not provided, uses CS_COPILOT_MEMORY_DB. Use unique paths for session isolation in testing.

 None
enable_mlflow_tracking bool

Enable MLflow tracking for agents (default: True). Set to False to disable tracking.

 True
run_context Any

Optional v2 workflow RunContext used to record validated specialist handoffs. When it is a real RunContext, the team's tool calls are routed through the execution kernel immediately; otherwise ensure_agno_session_run attaches the chat's run before the first message.

 None
execution_mode ExecutionMode | str | None

off, observe, or enforce. Defaults to CS_COPILOT_AGNO_EXECUTION (enforce when unset).

 None

Returns:

Name Type Description
Team Team

Configured Cs_copilot team

Raises:

Type Description
AgentCreationError

If one or more agents fail to initialize
Source code in src/cs_copilot/agents/teams.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def get_cs_copilot_agent_team(
    model: Model,  # Agno Model instance, e.g. OpenAIChat(...) or Claude(...)
    *,
    markdown: bool = True,
    debug_mode: bool = False,
    show_members_responses: bool = True,
    enable_memory: bool = True,
    db_file: str = None,
    enable_mlflow_tracking: bool = True,
    run_context: Any = None,
    execution_mode: ExecutionMode | str | None = None,
) -> Team:
    """
    Create a coordinated team of cs_copilot agents using Agno.

    Args:
        model: Agno Model instance used for team coordination and member agents
        markdown: Format output in markdown
        debug_mode: Enable debug logs
        show_members_responses: Print member responses during coordination
        enable_memory: Enable persistent session history (default: True). Cross-session
                      user/agentic memories stay disabled to prevent state leakage.
        db_file: Custom database file path. If not provided, uses CS_COPILOT_MEMORY_DB.
                Use unique paths for session isolation in testing.
        enable_mlflow_tracking: Enable MLflow tracking for agents (default: True).
                               Set to False to disable tracking.
        run_context: Optional v2 workflow ``RunContext`` used to record validated
                     specialist handoffs. When it is a real ``RunContext``, the
                     team's tool calls are routed through the execution kernel
                     immediately; otherwise ``ensure_agno_session_run`` attaches
                     the chat's run before the first message.
        execution_mode: ``off``, ``observe``, or ``enforce``. Defaults to
                     ``CS_COPILOT_AGNO_EXECUTION`` (``enforce`` when unset).

    Returns:
        Team: Configured Cs_copilot team

    Raises:
        AgentCreationError: If one or more agents fail to initialize
    """
    logger = logging.getLogger(__name__)
    logger.info("Creating Cs_copilot Agent Team")

    # ✅ Single DB handles session storage/history in v2.1.x.
    # Cross-session memories are intentionally disabled below; only per-thread
    # history/session state should persist.
    db = None
    if enable_memory:
        db = SqliteDb(
            db_file=db_file
            or CS_COPILOT_MEMORY_DB
            # NOTE: CS_COPILOT_MEMORY_TABLE is not required by SqliteDb.
            # Agno manages its own tables for sessions/memories. Kept import for compat.
        )

    # Probe runtime environment (GPU, CPU, RAM, databases, cached models)
    resource_profile = analyze_resources()
    logger.info("Resource profile: %s", resource_profile)
    coordinator_tools = [SessionMemoryToolkit(), SkillToolkit()]
    validate_role_tools(get_role_policy(COORDINATOR_ROLE), coordinator_tools)
    agentic_contracts = {
        "handoff_schema_version": HANDOFF_SCHEMA_VERSION,
        "context_budget": DEFAULT_CONTEXT_BUDGET.as_dict(),
        "delegation_limits": DEFAULT_DELEGATION_LIMITS.as_dict(),
        "handoff_transport": "delegate_task_to_member.task_description JSON",
        "role_profiles": {
            role: policy.profile for role, policy in ROLE_POLICIES.items() if role != "single_agent"
        },
    }
    active_workflow_run = getattr(run_context, "run", None)
    if active_workflow_run is not None:
        active_run = {
            "run_id": active_workflow_run.run_id,
            "workflow_slug": active_workflow_run.workflow_slug,
            "trace_id": active_workflow_run.trace_id,
        }
    else:
        # Even ad-hoc Agno sessions need canonical handoff identity. This is
        # deliberately process-local unless a durable RunContext is supplied.
        active_run = {
            "run_id": f"agno-{uuid.uuid4().hex[:12]}",
            "workflow_slug": "ad-hoc",
            "trace_id": uuid.uuid4().hex,
        }
    agentic_contracts["active_run"] = active_run
    shared_session_state = {
        "resource_profile": resource_profile,
        "agent_scratch": {},
        "agentic_contracts": agentic_contracts,
        "current_run_id": active_run["run_id"],
    }

    # Common agent parameters supplied by the factory
    agent_params = {
        "markdown": markdown,
        "debug_mode": debug_mode,
        "enable_mlflow_tracking": enable_mlflow_tracking,
        "session_state": shared_session_state,
    }

    # ============================================================================
    # 7-AGENT PRODUCTION TEAM
    # ============================================================================
    # Consolidation history:
    #   MERGED: GTM Optimization + Loading + Density + Activity → GTM Agent
    #   GENERALIZED: GTM Chemotype Analysis → Chemoinformatician (method-agnostic)
    #   TRANSFORMED: Autoencoder public agent → Molecular Designer (engine-based)
    #   ADDED: Report Generator (presentation layer)
    #   REMOVED: Robustness Evaluator (not included in main team, invoked separately)
    # ============================================================================

    # (type_key, human_name)
    agents_config: List[Tuple[str, str]] = [
        ("chembl_downloader", "ChEMBL Downloader"),
        (
            "gtm_agent",
            "GTM Agent",
        ),  # Unified GTM operations (build, load, density, activity, project)
        (
            "chemoinformatician",
            "Chemoinformatician",
        ),  # Comprehensive chemoinformatics (chemotype, clustering, SAR, similarity, QSAR)
        ("report_generator", "Report Generator"),  # Universal presentation layer
        ("molecular_designer", "Molecular Designer"),  # Small-molecule design engines
        ("peptide_designer", "Peptide Designer"),  # Peptide design engines
        ("synplanner", "SynPlanner"),
        # Note: Robustness Evaluator excluded from main team (invoked separately for testing)
    ]

    agents = []
    failures = []

    for agent_type, agent_name in agents_config:
        try:
            logger.info("Creating %s agent", agent_name)
            agent = create_agent(agent_type, model=model, **agent_params)
            agent.agentic_role = agent_type
            agents.append(agent)
            logger.info("Successfully created %s agent", agent_name)
        except Exception as e:
            logger.exception("Failed to create %s agent", agent_name)
            failures.append(f"{agent_name}: {e!s}")

    if failures:
        msg = "Agent initialization failures:\n  - " + "\n  - ".join(failures)
        raise AgentCreationError(msg)

    agentic_contracts["member_roles"] = {
        get_member_id(agent): agent.agentic_role for agent in agents
    }
    delegation_guard = StructuredDelegationGuard()
    team = StructuredHandoffTeam(
        name="Cs_copilot Team",
        members=agents,
        model=model,
        run_context=run_context,
        delegation_guard=delegation_guard,
        # ✅ Attach DB directly to the team (persists sessions/history)
        # If enable_memory=False, db=None prevents any persistence
        db=db,
        # Keep session history, but never inject cross-session memories. Agno
        # defaults add_memories_to_context=True when agentic memory is enabled,
        # which caused new chats to recall prior chemical-space analyses.
        enable_agentic_memory=False,
        enable_user_memories=False,
        add_memories_to_context=False,
        add_history_to_context=enable_memory,  # include recent history in prompts
        num_history_runs=5 if enable_memory else 0,  # 🔧 LIMIT context to last 5 runs
        # The coordinator passes bounded handoff envelopes; member transcripts
        # are not broadcast to peers as an implicit second context channel.
        share_member_interactions=False,
        store_history_messages=enable_memory,  # persist message history to DB
        store_tool_messages=enable_memory,  # persist tool results
        store_media=enable_memory,  # persist any media if used
        # Session state (always enabled for within-session data passing)
        session_state=shared_session_state,
        add_session_state_to_context=True,
        enable_agentic_state=True,
        tools=coordinator_tools,
        # Prompting
        description=(
            "You are an intelligent coordinator orchestrating a team of specialized cheminformatics agents. "
            "Your role is to understand user requests, select the appropriate agent(s) or workflows, "
            "and chain multiple agents when needed to complete complex analyses.\n\n"
            "• ChEMBL Downloader: Download bioactivity data from ChEMBL database\n"
            "• GTM Agent: All GTM operations (build/load/density/activity/project) with smart caching\n"
            "• Chemoinformatician: Downstream analysis (scaffold, SAR, similarity, clustering) - works with GTM output\n"
            "• Report Generator: Universal presentation layer for all analysis types\n"
            "• Molecular Designer: Small-molecule design via autoencoder and LLM engines (SMILES, standalone + GTM-guided)\n"
            "• Peptide Designer: Peptide design via WAE and LLM engines + GTM on latent space + DBAASP antimicrobial activity landscapes\n"
            "• SynPlanner: Retrosynthetic planning for target molecules\n\n"
            # Routing prose is generated from the workflow/skill catalog keywords
            # so it can never drift from the deterministic MCP bootstrap routing.
            + render_routing_rules() + "\n\n"
            "When coordinating: (1) Assess if a predefined workflow covers the request, (2) Select and chain "
            "specialized agents for multi-step tasks (GTM → Chemoinformatician → Report Generator is common), "
            "(3) For analysis requests, automatically add Report Generator unless user explicitly requests raw data only, "
            "(4) For ambiguous opening requests, apply the INITIAL CLARIFICATION FLOW (peptides vs molecules, then exploratory vs generative), (5) Synthesize insights from agent outputs into coherent analyses."
        ),
        instructions=[
            *AGENT_TEAM_INSTRUCTIONS,
            (
                "For multi-step scientific workflows, consult the Skills tools "
                "(`list_skills`, `search_skills`, `fetch_skill`) before routing "
                "specialized agents so the team follows reusable ChemSpace procedures."
            ),
            (
                "When calling `delegate_task_to_member`, `task_description` MUST be exactly "
                "one JSON object with schema_version, run_id, workflow_slug, task_id, "
                "sender_role, receiver_role, objective, constraints, required_capabilities, "
                "input_artifact_ids, expected_output_artifacts, expected_output_schema, "
                "acceptance_criteria, context_summary, budget (max_tokens, max_tool_calls, "
                "timeout_seconds), trace_id, span_id, and optional parent_span_id. Use the "
                "run_id, workflow_slug, and trace_id from agentic_contracts.active_run. Use the "
                "selected member's canonical role from agentic_contracts.member_roles as "
                "receiver_role. Never include private reasoning, scratchpads, messages, or "
                "conversation history. The delegation guard derives expected_output from "
                "this object and rejects unstructured, oversized, repeated, or looping calls."
            ),
        ],
        # UX & observability
        markdown=markdown,
        debug_mode=debug_mode,
        stream_member_events=True,  # stream events from members (Team API)
        show_members_responses=show_members_responses,
    )
    team.execution_mode = execution_mode_from_env(execution_mode)
    if (
        team.execution_mode is not ExecutionMode.OFF
        and hasattr(run_context, "bind_session_state")
        and hasattr(run_context, "layout")
    ):
        run_context.bind_session_state(shared_session_state)
        attach_execution(team, run_context=run_context, mode=team.execution_mode)
    logger.info("Successfully created Cs_copilot Agent Team")
    return team

utils

Utility functions for agent operations.

get_last_agent_reply(agent)

Extract the content of the last message from an agent's session.

Source code in src/cs_copilot/agents/utils.py 
12
13
14
def get_last_agent_reply(agent: Agent) -> str:
    """Extract the content of the last message from an agent's session."""
    return copy.deepcopy(agent.get_messages_for_session()[-1].to_dict()["content"])