When an upstream API returns structured JSON, a good pattern is:
- validate the response with a Pydantic model
- map that model into a
MaltegoEntitysubclass - return only entities from the transform
This keeps API validation separate from graph-building logic and makes the mapping code easier to test.
Recommended flow
Boundary validation
Use Pydantic at the HTTP boundary so invalid payloads fail early:
from pydantic import BaseModel
class RepoRecord(BaseModel):
name: str
html_url: str
description: str | None = NoneEntity definition
Keep the entity focused on graph-facing fields:
from maltego.server import (
MaltegoEntity,
MaltegoEntityConfig,
MaltegoEntityProperty as MEF,
register_entity,
)
@register_entity
class Repository(MaltegoEntity):
Config = MaltegoEntityConfig(
value_property="name",
display_name="Repository",
icon_resource="Website",
)
name: str = MEF(display_name="Name", sample_value="maltego-transforms")
html_url: str = MEF(
display_name="URL",
sample_value="https://example.com/repo",
)
description: str = MEF(
display_name="Description",
sample_value="Repository description",
)Mapper function
Put the translation logic in a small mapper. This is also a good place to apply entity features that should always travel with that result type. See the Entity Features (Overlays, Links, Notes) article for the broader set of options you can use to customize entity output.
from html import escape
from maltego.model.types import LinkColor, LinkStyle, LinkThickness
def repo_to_entity(repo: RepoRecord) -> Repository:
entity = Repository(repo.name)
entity.html_url = repo.html_url
entity.link_label = "repository"
entity.link_style = LinkStyle.DASHED
entity.link_color = LinkColor.BLUE
entity.link_thickness = LinkThickness.THICKNESS_2
entity.add_display_field_html(
"Repository URL",
f'<a href="{escape(repo.html_url)}">{escape(repo.name)}</a>',
)
if repo.description:
entity.description = repo.description
entity.note = repo.description
entity.add_display_field_markdown(
"Repository summary",
repo.description,
)
return entityTransform code
The transform should orchestrate the API call and mapping, not embed the schema translation inline:
from maltego.entities import Phrase
from maltego.model.context import MaltegoContext
from maltego.server import IntegrationClient, register_transform
client = IntegrationClient()
@register_transform
async def list_repositories(
input_entity: Phrase,
context: MaltegoContext,
) -> list[Repository]:
response = await client.get(
"https://api.example.com/repos",
context=context,
)
payload = response.json()
repos = [RepoRecord.model_validate(item) for item in payload["results"]]
return [repo_to_entity(repo) for repo in repos]Use the mapper for per-entity presentation and default link styling. When you need to control the graph structure itself, such as adding explicit links with context.graph or building a custom graph output, keep that logic in the transform.
Where this fits in a transform server project
In a project created with maltego-transforms start, keep the registered transform in transforms/ and move the upstream schemas and mapper functions into helper modules once they grow past a few fields. A common split is:
transforms/<feature>.pyfor the registered transformmodels.pyorapi.pyfor Pydantic models and HTTP client codemappers.pyformodel_to_entity()helpers
Best practices
- Validate external data once, as close to the HTTP response as possible.
- Keep Pydantic models and
MaltegoEntityclasses separate instead of trying to make one class do both jobs. - Use explicit mapper functions so field renames and nested transformations stay obvious.
- Use mapper functions to attach entity-level presentation such as notes, display fields, or default link styling; see the Entity Features (Overlays, Links, Notes) article.
- Return entities or graphs from the transform; keep raw API models out of the graph layer.
- When the upstream schema is nested, combine this pattern with the approach in the Composed Entities article.